package com.alibaba.otter.canal.utils;

import com.alibaba.otter.canal.constants.ModuleEnum;
import com.alibaba.otter.canal.custom.CanalLogUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanMap;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
 
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

/**
 * @author kaikai_zheng
 * 
 
@version 1.0.0
 * @className ESUtil
 * @description //ES工具類
 * @data 2020-07-13 11:19
 */
@Slf4j
public class ESUtil {

     // 改為從配置中讀取叢集節點
//    private final static String HOST_IP1 = "47.101.179.162";
//    private final static String HOST_IP2 = "10.2.6.58";
//    private final static String HOST_IP3 = "10.2.6.59";
 
//    private final static String HOST_IP4 = "10.2.6.60";
//    private final static int PORT = 9200;
    /**
     * 超時時間設為5分鐘
     */
    private static final int TIME_OUT = 5 * 60 * 1000;
    private static final int ADDRESS_LENGTH = 2;
    private static final String HTTP_SCHEME = "http";

//
//    private volatile static ESUtil esUtil;
//
    public ESUtil(){}

    // 雙重校驗鎖
//    public static ESUtil getInstance() {
//        if (null == esUtil) {
//            synchronized (ESUtil.class){
//                if (null == esUtil) {
//                    esUtil = new ESUtil();
//                }
//            }
//        }
//        return esUtil;
//    }

    public RestHighLevelClient getClient() {
        RestHighLevelClient client = new RestHighLevelClient(
                restClientBuilder()
//                RestClient.builder(
//                new HttpHost(HOST_IP1, PORT, PROTOCAL)
//                new HttpHost(HOST_IP2, PORT, PROTOCAL),
//                new HttpHost(HOST_IP3, PORT, PROTOCAL),
//                new HttpHost(HOST_IP4, PORT, PROTOCAL)
        );
        return client;
    }


    public void closeClient(RestHighLevelClient client) {
        try {
            client.close();
        } catch (IOException e) {
            CanalLogUtil.error(ModuleEnum.CANAL_CLIENT.getCode(), "colse es client exception,error=", e);
            e.printStackTrace();
        }
    }

    public RestClientBuilder restClientBuilder() {
        String ipAddress[] = ResourceUtil.getResource().getProperty("es_cluster").split(",");
        CanalLogUtil.info(ModuleEnum.CANAL_CLIENT.getCode(),"load es cluster nodes result="+ipAddress);
        HttpHost[] hosts = Arrays.stream(ipAddress)
                .map(this::makeHttpHost)
                .filter(Objects::nonNull)
                .toArray(HttpHost[]::new);
        return RestClient.builder(hosts);
    }

    private HttpHost makeHttpHost(String s) {
        assert StringUtils.isNotEmpty(s);
        String[] address = s.split(":");
        if (address.length == ADDRESS_LENGTH) {
            String ip = address[0];
            int port = Integer.parseInt(address[1]);
            System.err.println(ip+"+"+port);
            return new HttpHost(ip, port, HTTP_SCHEME);
        } else {
            return null;
        }
    }

    public static <T> Map<String, Object> beanToMap(T bean) {
        Map<String, Object> map = new HashMap<>();
        if (bean != null) {
            BeanMap beanMap = new BeanMap(bean);
            for (Object key : beanMap.keySet()) {
                if (beanMap.get(key) != null) {
                    map.put(key + "", beanMap.get(key));
                }
            }
        }
        return map;
    }

}