package com.zzsn.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;

import java.io.IOException;

/**
 * @author kongliufeng
 * @create 2020-07-31 17:40onfi
 */
@Slf4j
@Configuration
public class ElasticsearchClientConfig implements DisposableBean {

    @Value("${zzsn.es.host}")
    private String host;
    @Value("${zzsn.es.port}")
    private int port;
    @Value("${zzsn.es.url}")
    private String url;
    @Value("${zzsn.es.userName}")
    private String userName;
    @Value("${zzsn.es.passWord}")
    private String passWord;
    @Value("${zzsn.es.clusterIsStart}")
    private Boolean clusterIsStart;

    private final String scheme = "http";
    private final int connectTimeOut = 5000;
    private final int socketTimeOut = 60000;
    private final int connectionRequestTimeOut = 1500;
    private final int maxConnectPerRoute = 10;

    //es 集群node系欸但数量
    private final int nodesNum = 3;
    //最大连接数量
    private final int maxConnectNum = maxConnectPerRoute * nodesNum;

    private final boolean uniqueConnectTimeConfig = true;
    private final boolean uniqueConnectNumConfig = true;
    private RestClientBuilder builder;
    private RestHighLevelClient client;

    public ElasticsearchClientConfig() {
    }

    /*
    * 创建 RestHigLevelClient,@Scope("prototype")
    * */
    @Bean()
    public RestHighLevelClient restHighLevelClient(){
        System.out.println("生成bean：restHighLevelClient");
        //HttpHost httpHost = new HttpHost(host, port, scheme);
        if(clusterIsStart){
            HttpHost[] httpHosts = null;
            if(url!=null){
                String[] urls = url.split(",");
                httpHosts = new HttpHost[urls.length];
                for (int i = 0; i < urls.length; i++) {
                    String[] ipAndPort = urls[i].split(":");
                    httpHosts[i] = new HttpHost(ipAndPort[0],Integer.valueOf(ipAndPort[1]),scheme);
                }
            }
            builder = RestClient.builder(httpHosts);
        }else{
            HttpHost httpHost = new HttpHost(host, port, scheme);
            builder = RestClient.builder(httpHost);
        }

        if(uniqueConnectNumConfig){
            setConnectTimeOutConfig();
        }if (uniqueConnectNumConfig) {
            setMutiConnectConfig();
        }
       /*RestHighLevelClient restHighLevelClient = new RestHighLevelClient(
                RestClient.builder(httpHost)
        );*/
        client = new RestHighLevelClient(builder);
        return client;
    }

    private void setMutiConnectConfig() {
        builder.setHttpClientConfigCallback(HttpClientBuilder ->{
            //if(!("zzsn".equals(userName) && (Objects.isNull(passWord) || "zzsn9988".equals(passWord)))){
            if(!StringUtils.isEmpty(userName) && !StringUtils.isEmpty(passWord)){
                log.info("es集群服务器账号[{}],密码[{}]",userName,passWord);
                CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                credentialsProvider.setCredentials(AuthScope.ANY,
                        new UsernamePasswordCredentials(userName,passWord));
                HttpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            };
            return HttpClientBuilder.setMaxConnTotal(maxConnectNum)
                    .setMaxConnPerRoute(maxConnectPerRoute)
                    .setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(
                            Runtime.getRuntime().availableProcessors() << 1
                    ).build());
        });
    }

    /**
     * lambda表达式创建
     * 异步httpclient 的连接延时配置
     */
    private void setConnectTimeOutConfig() {
        builder.setRequestConfigCallback(requestConfigBuilder ->
                requestConfigBuilder
                        .setConnectTimeout(connectTimeOut)
                        .setSocketTimeout(socketTimeOut)
                        .setConnectionRequestTimeout(connectionRequestTimeOut)
        );
    }

    /**
     * 销毁方法
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
        close();
    }

    public void close() {
        if (client != null) {
            try {
                client.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}
