Java 类org.apache.http.impl.nio.reactor.IOReactorConfig 实例源码

private CloseableHttpAsyncClient createHttpAsyncClient(YunpianConf conf) throws IOReactorException {
    IOReactorConfig ioReactorConfig = IOReactorConfig.custom().setIoThreadCount(Runtime.getRuntime().availableProcessors())
            .setConnectTimeout(conf.getConfInt(YunpianConf.HTTP_CONN_TIMEOUT, "10000"))
            .setSoTimeout(conf.getConfInt(YunpianConf.HTTP_SO_TIMEOUT, "30000")).build();
    ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);

    PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(ioReactor);
    ConnectionConfig connectionConfig = ConnectionConfig.custom().setMalformedInputAction(CodingErrorAction.IGNORE)
            .setUnmappableInputAction(CodingErrorAction.IGNORE)
            .setCharset(Charset.forName(conf.getConf(YunpianConf.HTTP_CHARSET, YunpianConf.HTTP_CHARSET_DEFAULT))).build();
    connManager.setDefaultConnectionConfig(connectionConfig);
    connManager.setMaxTotal(conf.getConfInt(YunpianConf.HTTP_CONN_MAXTOTAL, "100"));
    connManager.setDefaultMaxPerRoute(conf.getConfInt(YunpianConf.HTTP_CONN_MAXPERROUTE, "10"));

    CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setConnectionManager(connManager).build();
    httpclient.start();
    return httpclient;
}

public FiberApacheHttpClientRequestExecutor(final Validator<CloseableHttpResponse> resValidator, final int maxConnections, final int timeout, final int parallelism) throws IOReactorException {
  final DefaultConnectingIOReactor ioreactor = new DefaultConnectingIOReactor(IOReactorConfig.custom().
    setConnectTimeout(timeout).
    setIoThreadCount(parallelism).
    setSoTimeout(timeout).
    build());

  final PoolingNHttpClientConnectionManager mngr = new PoolingNHttpClientConnectionManager(ioreactor);
  mngr.setDefaultMaxPerRoute(maxConnections);
  mngr.setMaxTotal(maxConnections);

  final CloseableHttpAsyncClient ahc = HttpAsyncClientBuilder.create().
    setConnectionManager(mngr).
    setDefaultRequestConfig(RequestConfig.custom().setLocalAddress(null).build()).build();

  client = new FiberHttpClient(ahc);
  validator = resValidator;
}

/**
 * Creates asynchronous Apache HTTP client.
 *
 * @param settings
 *         settings to use to create client.
 * @param conf
 *         configuration related to async connection.
 * @return Instance of {@link CloseableHttpAsyncClient}.
 */
private CloseableHttpAsyncClient createClient(HttpSettings settings, ApacheHttpClientConfiguration conf) {
    IOReactorConfig ioReactor = IOReactorConfig.custom().setIoThreadCount(conf.getMaxThreadCount()).build();
    HttpAsyncClientBuilder httpClientBuilder = HttpAsyncClients.custom()
            .useSystemProperties()
                    // allow POST redirects
            .setRedirectStrategy(new LaxRedirectStrategy()).setMaxConnTotal(conf.getMaxTotalConnectionCount()).setMaxConnPerRoute(conf.getMaxRouteConnectionCount()).setDefaultIOReactorConfig(ioReactor)
            .setKeepAliveStrategy(new DefaultConnectionKeepAliveStrategy()).setDefaultRequestConfig(createDefaultRequestConfig(settings));
    if (settings.getProxyUrl() != null) {
        DefaultProxyRoutePlanner routePlanner = createProxyRoutePlanner(settings, httpClientBuilder);
        httpClientBuilder.setRoutePlanner(routePlanner);
    }
    CloseableHttpAsyncClient httpClient = httpClientBuilder.build();
    httpClient.start();
    return httpClient;
}

@Bean
public CloseableHttpAsyncClient asyncHttpClient() {
    try {
        PoolingNHttpClientConnectionManager connectionManager = new PoolingNHttpClientConnectionManager(
                new DefaultConnectingIOReactor(IOReactorConfig.DEFAULT));
        connectionManager.setMaxTotal(DEFAULT_MAX_TOTAL_CONNECTIONS);
        connectionManager.setDefaultMaxPerRoute(DEFAULT_MAX_CONNECTIONS_PER_ROUTE);
        RequestConfig config = RequestConfig.custom()
                .setConnectTimeout(DEFAULT_READ_TIMEOUT_MILLISECONDS)
                .build();

        return HttpAsyncClientBuilder.create()
                .setConnectionManager(connectionManager)
                .setDefaultRequestConfig(config)
                .build();
    } catch (Exception e) {
        throw Throwables.propagate(e);
    }
}

public SleepServerApiClient() throws Exception {
    connectionManager = new PoolingNHttpClientConnectionManager(
            new DefaultConnectingIOReactor(IOReactorConfig.DEFAULT));
    connectionManager.setMaxTotal(20000);
    connectionManager.setDefaultMaxPerRoute(20000);

    RequestConfig config = RequestConfig.custom().setConnectTimeout(120000)
            .build();

    CloseableHttpAsyncClient httpClient = HttpAsyncClientBuilder.create()
            .setConnectionManager(connectionManager)
            .setDefaultRequestConfig(config).build();

    HttpComponentsAsyncClientHttpRequestFactory requestFactory = new HttpComponentsAsyncClientHttpRequestFactory(
            httpClient);
    client = new AsyncRestTemplate(requestFactory);
}

public void start() throws Exception {
    // Create HTTP protocol basic processing chain
    HttpProcessor httpProcessor = HttpProcessorBuilder.create()
            .add(new ResponseDate()).add(new ResponseContent())
            .add(new ResponseConnControl()).build();
    // Create server
    HttpAsyncService protocolHandler = new HttpAsyncService(httpProcessor,
            uriMapper);
    NHttpConnectionFactory<DefaultNHttpServerConnection> connFactory = new DefaultNHttpServerConnectionFactory();
    IOEventDispatch ioEventDispatch = new DefaultHttpServerIODispatch(
            protocolHandler, connFactory);
    IOReactorConfig config = IOReactorConfig.custom()
            .setIoThreadCount(threads).setSoReuseAddress(true).build();
    ListeningIOReactor ioReactor = new DefaultListeningIOReactor(config);
    // Start server
    ioReactor.listen(new InetSocketAddress(port));
    ioReactor.execute(ioEventDispatch);
}

/**
 * Build the {@link RestClient} used for reindexing from remote clusters.
 * @param remoteInfo connection information for the remote cluster
 * @param taskId the id of the current task. This is added to the thread name for easier tracking
 * @param threadCollector a list in which we collect all the threads created by the client
 */
static RestClient buildRestClient(RemoteInfo remoteInfo, long taskId, List<Thread> threadCollector) {
    Header[] clientHeaders = new Header[remoteInfo.getHeaders().size()];
    int i = 0;
    for (Map.Entry<String, String> header : remoteInfo.getHeaders().entrySet()) {
        clientHeaders[i] = new BasicHeader(header.getKey(), header.getValue());
    }
    return RestClient.builder(new HttpHost(remoteInfo.getHost(), remoteInfo.getPort(), remoteInfo.getScheme()))
            .setDefaultHeaders(clientHeaders)
            .setRequestConfigCallback(c -> {
                c.setConnectTimeout(Math.toIntExact(remoteInfo.getConnectTimeout().millis()));
                c.setSocketTimeout(Math.toIntExact(remoteInfo.getSocketTimeout().millis()));
                return c;
            })
            .setHttpClientConfigCallback(c -> {
                // Enable basic auth if it is configured
                if (remoteInfo.getUsername() != null) {
                    UsernamePasswordCredentials creds = new UsernamePasswordCredentials(remoteInfo.getUsername(),
                            remoteInfo.getPassword());
                    CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                    credentialsProvider.setCredentials(AuthScope.ANY, creds);
                    c.setDefaultCredentialsProvider(credentialsProvider);
                }
                // Stick the task id in the thread name so we can track down tasks from stack traces
                AtomicInteger threads = new AtomicInteger();
                c.setThreadFactory(r -> {
                    String name = "es-client-" + taskId + "-" + threads.getAndIncrement();
                    Thread t = new Thread(r, name);
                    threadCollector.add(t);
                    return t;
                });
                // Limit ourselves to one reactor thread because for now the search process is single threaded.
                c.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(1).build());
                return c;
            }).build();
}

@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;
}

public static void main(String[] args) throws InterruptedException, IOException
{
   try
   {
      IOReactorConfig config = IOReactorConfig.custom().setSoTimeout(15000).setTcpNoDelay(true).build();
      final HttpServer server = ServerBootstrap.bootstrap().setListenerPort(PORT).setServerInfo("Test/1.1").setIOReactorConfig(config).setExceptionLogger(ExceptionLogger.STD_ERR).registerHandler("*", new HTTPTimeHandler()).create();

      server.start();

      System.out.println("Server started");

      Runtime.getRuntime().addShutdownHook(new Thread()
      {
         @Override public void run()
         {
            System.out.println("Server shutdown requested");
            server.shutdown(5, TimeUnit.SECONDS);
         }
      });

      server.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
   }
   finally
   {
      System.out.println("Server shutdown");
   }
}

public AsyncServiceClient(ClientConfiguration config) {
    super(config);
    try {
        IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
                .setIoThreadCount(config.getIoThreadCount()).build();
        ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(
                ioReactorConfig);
        PoolingNHttpClientConnectionManager cm = new PoolingNHttpClientConnectionManager(
                ioReactor);
        cm.setMaxTotal(config.getMaxConnections());
        cm.setDefaultMaxPerRoute(config.getMaxConnections());
        httpClient = new HttpFactory().createHttpAsyncClient(config, cm);

        /*
         * socketTimeout的值限制了closeIdleConnections执行的周期。
         * 如果周期相对socketTimeout的值过长，有可能一个请求分配到一个即将socketTimeout的连接，
         * 在请求发送之前即抛出SocketTimeoutException。
         * 现在让closeIdleConnections的执行周期为socketTimeout / 2.5。
         */
        long closePeriod = 5000;
        if (config.getSocketTimeoutInMillisecond() > 0) {
            closePeriod = (long) (config.getSocketTimeoutInMillisecond() / 2.5);
        }
        closePeriod = closePeriod < 5000 ? closePeriod : 5000;
        connEvictor = new IdleConnectionEvictor(cm, closePeriod);
        httpClient.start();
        connEvictor.start();
    } catch (IOReactorException ex) {
        throw new ClientException(String.format("IOReactorError: %s",
                ex.getMessage()), ex);
    }
}

@Bean
PoolingNHttpClientConnectionManager poolingNHttpClientConnectionManager() throws IOReactorException {
    PoolingNHttpClientConnectionManager connectionManager = new PoolingNHttpClientConnectionManager(
            new DefaultConnectingIOReactor(IOReactorConfig.DEFAULT));
    connectionManager.setMaxTotal(maxTotalConnections);
    connectionManager.setDefaultMaxPerRoute(maxConnectionsPerRoute);
    return connectionManager;
}

@Bean
public CloseableHttpAsyncClient asyncHttpClient() throws Exception {
    PoolingNHttpClientConnectionManager connectionManager = new PoolingNHttpClientConnectionManager(new DefaultConnectingIOReactor(
            IOReactorConfig.DEFAULT));
    connectionManager.setMaxTotal(DEFAULT_MAX_TOTAL_CONNECTIONS);
    connectionManager.setDefaultMaxPerRoute(DEFAULT_MAX_CONNECTIONS_PER_ROUTE);
    connectionManager.setMaxPerRoute(new HttpRoute(new HttpHost("localhost")), 20);
    RequestConfig config = RequestConfig.custom().setConnectTimeout(DEFAULT_READ_TIMEOUT_MILLISECONDS).build();

    CloseableHttpAsyncClient httpclient = HttpAsyncClientBuilder.create().setConnectionManager(connectionManager).setDefaultRequestConfig(config)
            .build();
    return httpclient;
}

private HttpAsyncClientBuilder initialize() {
    try {
        final PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(
            new DefaultConnectingIOReactor( IOReactorConfig.custom()
                .setConnectTimeout( connectTimeout )
                .setSoTimeout( readTimeout )
                .build() ),
            RegistryBuilder.<SchemeIOSessionStrategy>create()
                .register( "http", NoopIOSessionStrategy.INSTANCE )
                .register( "https",
                    new SSLIOSessionStrategy( certificateLocation != null ?
                        createSSLContext( certificateLocation, certificatePassword )
                        : SSLContexts.createDefault(),
                        split( System.getProperty( "https.protocols" ) ),
                        split( System.getProperty( "https.cipherSuites" ) ),
                        new DefaultHostnameVerifier( PublicSuffixMatcherLoader.getDefault() ) ) )
                .build() );

        connManager.setMaxTotal( maxConnTotal );
        connManager.setDefaultMaxPerRoute( maxConnPerRoute );

        return ( certificateLocation != null ?
            HttpAsyncClients.custom()
                .setSSLContext( createSSLContext( certificateLocation, certificatePassword ) )
            : HttpAsyncClients.custom() )
            .setMaxConnPerRoute( maxConnPerRoute )
            .setConnectionManager( connManager )
            .setMaxConnTotal( maxConnTotal )
            .setKeepAliveStrategy( DefaultConnectionKeepAliveStrategy.INSTANCE )
            .setDefaultRequestConfig( RequestConfig
                .custom()
                .setRedirectsEnabled( redirectsEnabled )
                .setCookieSpec( cookieSpec )
                .build() )
            .setDefaultCookieStore( basicCookieStore );
    } catch( IOReactorException e ) {
        throw new UncheckedIOException( e );
    }
}

@SneakyThrows
public NioServer( int port, int workers ) {
    this.port = port;
    this.mapper.register( "/static/*", new NioClasspathResourceHandler( "/static", "/WEB-INF" ) );

    val ioReactorConfig = IOReactorConfig.custom().setIoThreadCount( workers ).build();
    val httpProcessor = HttpProcessorBuilder.create()
        .add( new ResponseDate() )
        .add( new ResponseServer( "OAP Server/1.0" ) )
        .add( new ResponseContent() )
        .add( new ResponseConnControl() )
        .build();

    SSLContext sslContext = getSslContext( port );


    server = ServerBootstrap.bootstrap()
        .setListenerPort( port )
        .setServerInfo( "OAP Server/1.0" )
        .setConnectionReuseStrategy( DefaultClientConnectionReuseStrategy.INSTANCE )
        .setHttpProcessor( httpProcessor )
        .setIOReactorConfig( ioReactorConfig )
        .setSslContext( sslContext )
        .setExceptionLogger( ex -> log.debug( ex.getMessage(), ex ) )
        .setHandlerMapper( mapper )
        .create();
}

@SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
public OppoMessenger(final String remoteHost, final int remotePort, final int localPort,
                     final HttpClientService httpClient)
        throws IOException {
    this.httpClient = httpClient;

    // Set up the server
    final HttpProcessor processor = HttpProcessorBuilder.create()
            .add(new ResponseContent())
            .add(new ResponseContentEncoding())
            .add(new ResponseConnControl())
            .build();

    final HttpAsyncService service = new HttpAsyncService(processor, mapper);
    final NHttpConnectionFactory<DefaultNHttpServerConnection> connectionFactory = new DefaultNHttpServerConnectionFactory(ConnectionConfig.DEFAULT);
    final IOEventDispatch dispatch = new DefaultHttpServerIODispatch(service, connectionFactory);

    server = new DefaultListeningIOReactor(IOReactorConfig.DEFAULT);
    server.listen(new InetSocketAddress(localPort));

    new Thread(new Runnable() {
        @Override public void run() {
            try {
                server.execute(dispatch);
            } catch (final IOException e) {
                logger().level(Error).message("HTTP server failed").error(e).log();
            }
        }
    }, "Oppo HTTP server");

    // Set up the client
    deviceUrlBase = "http://" + remoteHost + ':' + remotePort + '/';
}

@Override
public void init(ServletConfig config) {
    IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
            .setIoThreadCount(Parameters.HTTP_ASYNC_CLIENT_THREADS).build();
    httpClient = HttpAsyncClients.custom()
            .setDefaultIOReactorConfig(ioReactorConfig)
            .setMaxConnPerRoute(Parameters.HTTP_CLIENT_MAX_CONNECTIONS)
            .setMaxConnTotal(Parameters.HTTP_CLIENT_MAX_CONNECTIONS)
            .build();
    httpClient.start();
}

public TestRequestHandler() {
    IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
            .setIoThreadCount(Parameters.HTTP_ASYNC_CLIENT_THREADS).build();
    httpClient = HttpAsyncClients.custom()
            .setDefaultIOReactorConfig(ioReactorConfig)
            .setMaxConnPerRoute(Parameters.HTTP_CLIENT_MAX_CONNECTIONS)
            .setMaxConnTotal(Parameters.HTTP_CLIENT_MAX_CONNECTIONS).build();
    httpClient.start();
}

/**
 * Create connection manager for asynchronous http client.
 *
 * @return Connection manager for asynchronous http client.
 * @throws IOReactorException in case if a non-recoverable I/O error.
 */
protected NHttpClientConnectionManager createNHttpClientConnectionManager() throws IOReactorException {
    ConnectingIOReactor ioReactor =
            new DefaultConnectingIOReactor(IOReactorConfig.custom()
                    .setSoTimeout(this.config.getSocketTimeoutInMillis()).setTcpNoDelay(true).build());
    PoolingNHttpClientConnectionManager connectionManager = new PoolingNHttpClientConnectionManager(ioReactor);
    connectionManager.setDefaultMaxPerRoute(this.config.getMaxConnections());
    connectionManager.setMaxTotal(this.config.getMaxConnections());
    return connectionManager;
}

public AsyncClient() throws IOException {

        // Create I/O reactor configuration
        IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
                .setIoThreadCount(Runtime.getRuntime().availableProcessors())
                .setTcpNoDelay(true)
                .build();

        // Create a custom I/O reactor
        ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);

        // Create a custom Connection Manager
        connManager = new PoolingNHttpClientConnectionManager(ioReactor);
        connManager.setDefaultMaxPerRoute(defaultMaxPerRoute);
        connManager.setMaxTotal(defaultMaxPerRoute * MAX_CONN_MULTIPLICATION);

        // Create global request configuration
        RequestConfig defaultRequestConfig = RequestConfig.custom()
                .setCookieSpec(CookieSpecs.DEFAULT)
                .setExpectContinueEnabled(true)
                .setTargetPreferredAuthSchemes(Arrays.asList(AuthSchemes.NTLM, AuthSchemes.DIGEST))
                .setProxyPreferredAuthSchemes(Arrays.asList(AuthSchemes.BASIC))
                .build();

        // Update the request level configuration
        updateRequestConfig();

        // Create a custom Redirect Handler
        redirectStrategy = new RedirectHandler(requestConfig);

        httpAsyncClient = HttpAsyncClients.custom()
                .setConnectionManager(connManager)
                .setDefaultRequestConfig(defaultRequestConfig)
                .setRedirectStrategy(redirectStrategy)
                .build();
    }

private void initialize() {
    if (initialized.getAndSet(true)) {
        return;
    }
    IOReactorConfig.Builder config = createConfig();
    // params.setParameter(CoreProtocolPNames.USER_AGENT, "jsonrpc4j/1.0");
    final ConnectingIOReactor ioReactor = createIoReactor(config);
    createSslContext();
    int socketBufferSize = Integer.getInteger("com.googlecode.jsonrpc4j.async.socket.buffer", 8 * 1024);
    final ConnectionConfig connectionConfig = ConnectionConfig.custom().setBufferSize(socketBufferSize).build();
    BasicNIOConnFactory nioConnFactory = new BasicNIOConnFactory(sslContext, null, connectionConfig);
    pool = new BasicNIOConnPool(ioReactor, nioConnFactory, Integer.getInteger("com.googlecode.jsonrpc4j.async.connect.timeout", 30000));
    pool.setDefaultMaxPerRoute(Integer.getInteger("com.googlecode.jsonrpc4j.async.max.inflight.route", 500));
    pool.setMaxTotal(Integer.getInteger("com.googlecode.jsonrpc4j.async.max.inflight.total", 500));

    Thread t = new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                HttpAsyncRequestExecutor protocolHandler = new HttpAsyncRequestExecutor();
                IOEventDispatch ioEventDispatch = new DefaultHttpClientIODispatch(protocolHandler, sslContext, connectionConfig);
                ioReactor.execute(ioEventDispatch);
            } catch (InterruptedIOException ex) {
                System.err.println("Interrupted");
            } catch (IOException e) {
                System.err.println("I/O error: " + e.getMessage());
            }
        }
    }, "jsonrpc4j HTTP IOReactor");

    t.setDaemon(true);
    t.start();

    HttpProcessor httpProcessor = new ImmutableHttpProcessor(new RequestContent(), new RequestTargetHost(), new RequestConnControl(), new RequestUserAgent(), new RequestExpectContinue(false));
    requester = new HttpAsyncRequester(httpProcessor, new DefaultConnectionReuseStrategy());
}

private IOReactorConfig.Builder createConfig() {
    IOReactorConfig.Builder config = IOReactorConfig.custom();
    config = config.setSoTimeout(Integer.getInteger("com.googlecode.jsonrpc4j.async.socket.timeout", 30000));
    config = config.setConnectTimeout(Integer.getInteger("com.googlecode.jsonrpc4j.async.connect.timeout", 30000));
    config = config.setTcpNoDelay(Boolean.valueOf(System.getProperty("com.googlecode.jsonrpc4j.async.tcp.nodelay", "true")));
    config = config.setIoThreadCount(Integer.getInteger("com.googlecode.jsonrpc4j.async.reactor.threads", 1));
    return config;
}

protected ExecCallbackAsyncREST<HttpResponse> buildAsyncClient(RESTPool pool) throws IOException {
    SSLContext sslContext;
    try {
        sslContext = SSLContext.getDefault();
    } catch (NoSuchAlgorithmException e) {
        throw new IOException(e);
    }

    Registry<SchemeIOSessionStrategy> socketRegistry = RegistryBuilder.<SchemeIOSessionStrategy>create()
            .register("http", NoopIOSessionStrategy.INSTANCE)
            .register("https", new SSLIOSessionStrategy(sslContext, NoopHostnameVerifier.INSTANCE))
            .build();

    IOReactorConfig socketConfig = IOReactorConfig.custom()
            .setIoThreadCount(pool.getReactorThreadCount())
            .setSoTimeout(new Long(pool.getSocketTimeout()).intValue())
            .setTcpNoDelay(true)
            .setSoKeepAlive(true)
            .setSelectInterval(REACTOR_SELECT_INTERVAL)
            .build();

    ConnectionConfig connectionConfig = ConnectionConfig.custom()
            .setCharset(StandardCharsets.UTF_8)
            .setMalformedInputAction(CodingErrorAction.IGNORE)
            .setUnmappableInputAction(CodingErrorAction.IGNORE)
            .build();

    RequestConfig requestConfig = RequestConfig.custom()
            .setConnectionRequestTimeout(new Long(pool.getMaxPoolWait()).intValue())
            .setConnectTimeout(new Long(pool.getConnectionTimeout()).intValue())
            .setExpectContinueEnabled(pool.expectContinue())
            .setRedirectsEnabled(false)
            .setStaleConnectionCheckEnabled(pool.getValidationOnInactivity() >= 0)
            .build();

    NHttpConnectionFactory<ManagedNHttpClientConnection> connFactory = new ManagedNHttpClientConnectionFactory(
            new org.apache.http.impl.nio.codecs.DefaultHttpRequestWriterFactory(),
            new org.apache.http.impl.nio.codecs.DefaultHttpResponseParserFactory(),
            HeapByteBufferAllocator.INSTANCE
    );

    //TODO set validateAfterInactivity when supported
    PoolingNHttpClientConnectionManager ccm = new PoolingNHttpClientConnectionManager(
            new DefaultConnectingIOReactor(socketConfig),
            connFactory,
            socketRegistry,
            new SystemDefaultDnsResolver()
    );

    ccm.setMaxTotal(pool.getMaxTotal());
    ccm.setDefaultMaxPerRoute(pool.getMaxPerRoute());
    ccm.setDefaultConnectionConfig(connectionConfig);

    HttpAsyncClientBuilder builder = HttpAsyncClients.custom()
            .setConnectionManager(ccm)
            .setDefaultRequestConfig(requestConfig)
            .setKeepAliveStrategy(DefaultConnectionKeepAliveStrategy.INSTANCE)
            .disableCookieManagement();

    IdleAsyncConnectionEvictor evictor = new IdleAsyncConnectionEvictor(ccm, pool.getEvictorSleep(), TimeUnit.MILLISECONDS, pool.getMaxIdleTime(), TimeUnit.MILLISECONDS);

    addProxy(pool, builder);

    handleRedirects(pool, builder);

    CloseableHttpAsyncClient servClient = builder.build();

    servClient.start();

    HTTPCClientMonitor monitor = pool.hasConnectionMetrics() ? new HTTPCAsyncClientMonitor(pool.getName(), ccm) : null;

    return new HTTPCAsyncClient(servClient, evictor, monitor);
}

@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;
}