Java 类org.elasticsearch.client.sniff.Sniffer 实例源码

@Override
protected RestClient createInstance() throws Exception {
    HttpHost[] hosts = new HttpHost[hostnames.length];
    for (int i = 0; i < hosts.length; i++) {
        hosts[i] = HttpHost.create(hostnames[i]);
    }

    Header[] defaultHeaders = new Header[]{new BasicHeader(HEADER_CONTENT_TYPE_KEY, DEFAULT_HEADER_CONTENT_TYPE)};

    RestClient restClient = RestClient
            .builder(hosts)
            .setDefaultHeaders(defaultHeaders)
            .setFailureListener(loggingFailureListener)
            .build();

    if (enableSniffer) {
        this.sniffer = Sniffer.builder(restClient).build();
    }
    return restClient;
}

private void addSniffer(HttpHost[] hosts) {
  if (conf.clientSniff) {
    switch (hosts[0].getSchemeName()) {
      case "http":
        sniffer = Sniffer.builder(restClient).build();
        break;
      case "https":
        HostsSniffer hostsSniffer = new ElasticsearchHostsSniffer(
            restClient,
            ElasticsearchHostsSniffer.DEFAULT_SNIFF_REQUEST_TIMEOUT,
            ElasticsearchHostsSniffer.Scheme.HTTPS
        );
        sniffer = Sniffer.builder(restClient).setHostsSniffer(hostsSniffer).build();
        break;
      default:
        // unsupported scheme. do nothing.
    }
  }
}

@Inject
public EsRestClient(JkesProperties jkesProperties) {
    SniffOnFailureListener sniffOnFailureListener = new SniffOnFailureListener();
    Header[] defaultHeaders = {new BasicHeader("Content-Type", "application/json")};

    String[] urls = jkesProperties.getEsBootstrapServers().split("\\s*,");
    HttpHost[] hosts = new HttpHost[urls.length];
    for (int i = 0; i < urls.length; i++) {
        hosts[i] = HttpHost.create(urls[i]);
    }

    RestClient restClient = RestClient.builder(hosts)
            .setRequestConfigCallback(requestConfigBuilder -> {
                return requestConfigBuilder.setConnectTimeout(5000) // default 1s
                        .setSocketTimeout(60000); // defaults to 30 seconds
            }).setHttpClientConfigCallback(httpClientBuilder -> {
                return httpClientBuilder.setDefaultIOReactorConfig(
                        IOReactorConfig.custom().setIoThreadCount(2).build()); // because only used for admin, so not necessary to hold many worker threads
            })
            .setMaxRetryTimeoutMillis(60000) // defaults to 30 seconds
            .setDefaultHeaders(defaultHeaders)
            .setFailureListener(sniffOnFailureListener)
            .build();

    Sniffer sniffer = Sniffer.builder(restClient).build();
    sniffOnFailureListener.setSniffer(sniffer);

    this.sniffer = sniffer;
    this.restClient = restClient;
}

private void initESClient() throws NumberFormatException,
        UnknownHostException {

    List<HttpHost> httpHostList = hosts.stream().map(hostString -> {
        String[] parsedHost = hostString.split(":");
        String host = parsedHost[0];
        int port = parsedHost.length == 2 ? Integer.valueOf(parsedHost[1]) : 9200;
        return new HttpHost(host, port);
    }).collect(toList());
    List<HttpHost> clusterHosts = unmodifiableList(httpHostList);

    if (config.containsKey("user") && config.containsKey("password")) {
        String user = config.get("user").toString();
        String password = config.get("password").toString();
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials(user, password));

        restClient = RestClient.builder(clusterHosts.toArray(new HttpHost[clusterHosts.size()]))
                .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                        return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                    }
                }).build();
    } else {
        restClient = RestClient.builder(clusterHosts.toArray(new HttpHost[clusterHosts.size()]))
                .build();
    }
    if (this.isSniff) {
        sniffer = Sniffer.builder(restClient).build();
    }

}

@Autowired
public EsRestClient(JkesSearchProperties jkesProperties) {
    SniffOnFailureListener sniffOnFailureListener = new SniffOnFailureListener();
    Header[] defaultHeaders = {new BasicHeader("Content-Type", "application/json")};

    String[] urls = jkesProperties.getEs().getServers().split("\\s*,");
    HttpHost[] hosts = new HttpHost[urls.length];
    for (int i = 0; i < urls.length; i++) {
        hosts[i] = HttpHost.create(urls[i]);
    }

    RestClient restClient = RestClient.builder(hosts)
            .setRequestConfigCallback(requestConfigBuilder -> {
                return requestConfigBuilder.setConnectTimeout(5000) // default 1s
                        .setSocketTimeout(60000); // defaults to 30 seconds
            }).setHttpClientConfigCallback(httpClientBuilder -> {
                return httpClientBuilder.setDefaultIOReactorConfig(
                        IOReactorConfig.custom().setIoThreadCount(2).build()); // because only used for admin, so not necessary to hold many worker threads
            })
            .setMaxRetryTimeoutMillis(60000) // defaults to 30 seconds
            .setDefaultHeaders(defaultHeaders)
            .setFailureListener(sniffOnFailureListener)
            .build();

    Sniffer sniffer = Sniffer.builder(restClient).build();
    sniffOnFailureListener.setSniffer(sniffer);

    this.sniffer = sniffer;
    this.restClient = restClient;
}