Java 类org.apache.http.impl.nio.client.HttpAsyncClientBuilder 实例源码

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

private RemoteScrollableHitSource sourceWithMockedClient(boolean mockRemoteVersion, CloseableHttpAsyncClient httpClient)
        throws Exception {
    HttpAsyncClientBuilder clientBuilder = mock(HttpAsyncClientBuilder.class);
    when(clientBuilder.build()).thenReturn(httpClient);

    RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200))
            .setHttpClientConfigCallback(httpClientBuilder -> clientBuilder).build();

    TestRemoteScrollableHitSource hitSource = new TestRemoteScrollableHitSource(restClient) {
        @Override
        void lookupRemoteVersion(Consumer<Version> onVersion) {
            if (mockRemoteVersion) {
                onVersion.accept(Version.CURRENT);
            } else {
                super.lookupRemoteVersion(onVersion);
            }
        }
    };
    if (mockRemoteVersion) {
        hitSource.remoteVersion = Version.CURRENT;
    }
    return hitSource;
}

private CloseableHttpAsyncClient createHttpClient() {
    //default timeouts are all infinite
    RequestConfig.Builder requestConfigBuilder = RequestConfig.custom()
            .setConnectTimeout(DEFAULT_CONNECT_TIMEOUT_MILLIS)
            .setSocketTimeout(DEFAULT_SOCKET_TIMEOUT_MILLIS)
            .setConnectionRequestTimeout(DEFAULT_CONNECTION_REQUEST_TIMEOUT_MILLIS);
    if (requestConfigCallback != null) {
        requestConfigBuilder = requestConfigCallback.customizeRequestConfig(requestConfigBuilder);
    }

    HttpAsyncClientBuilder httpClientBuilder = HttpAsyncClientBuilder.create().setDefaultRequestConfig(requestConfigBuilder.build())
            //default settings for connection pooling may be too constraining
            .setMaxConnPerRoute(DEFAULT_MAX_CONN_PER_ROUTE).setMaxConnTotal(DEFAULT_MAX_CONN_TOTAL);
    if (httpClientConfigCallback != null) {
        httpClientBuilder = httpClientConfigCallback.customizeHttpClient(httpClientBuilder);
    }
    return httpClientBuilder.build();
}

public ElasticBulkSender(String user, String password, HttpHost... hosts) {
    this.user = user;
    this.password = password;
    this.hosts = hosts;
    this.restClient = RestClient.builder(hosts)
            .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                @Override
                public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                    if (!Strings.isBlank(user)) {
                        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(user, password));
                        return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                    } else {
                        return httpClientBuilder;
                    }
                }
            })
            .build();
}

@Before
public void setUp() throws Exception {
    /**
     * 如果es集群安装了x-pack插件则以此种方式连接集群
     * 1. java客户端的方式是以tcp协议在9300端口上进行通信
     * 2. http客户端的方式是以http协议在9200端口上进行通信
     */
    Settings settings = Settings.builder(). put("xpack.security.user", "elastic:changeme").build();
    client = new PreBuiltXPackTransportClient(settings)
            .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300));
    elasticsearchTemplate = new ElasticsearchTemplate(client);
    final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
    credentialsProvider.setCredentials(AuthScope.ANY,
            new UsernamePasswordCredentials("elastic", "changeme"));
    restClient = RestClient.builder(new HttpHost("localhost",9200,"http"))
            .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                @Override
                public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                    return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                }
            }).build();
}

private CloseableHttpAsyncClient createCloseableHttpAsyncClient() throws Exception {
        HttpAsyncClientBuilder builder = HttpAsyncClientBuilder.create();
        builder.useSystemProperties();
        builder.setSSLContext(SSLContext.getDefault());
        builder.setConnectionReuseStrategy(DefaultConnectionReuseStrategy.INSTANCE);
        builder.setMaxConnPerRoute(2);
        builder.setMaxConnTotal(2);
        builder.setDefaultRequestConfig(RequestConfig
            .custom()
            .setConnectionRequestTimeout(1000)
            .setConnectTimeout(2000)
            .setSocketTimeout(2000)
        .build()
        );
//        builder.setHttpProcessor()
        CloseableHttpAsyncClient hc = builder.build();
        hc.start();
        return hc;
    }

private void open() {
   if (open) {
      // Ignore
      return;
   }

   final HttpAsyncClientBuilder builder = HttpAsyncClients.custom()
         .setMaxConnPerRoute(1000)
         .setMaxConnTotal(1000)
         .setKeepAliveStrategy(new DefaultConnectionKeepAliveStrategy())
         .setThreadFactory(threadFactory);
   if (credentialsProvider != null) {
      builder.setDefaultCredentialsProvider(credentialsProvider);
   }

   httpclient = builder.build();
   httpclient.start();
   this.open = true;
}

@Test
public void defaultSettingsOfHttpAsyncClientLostOnExecutorCustomization() throws Exception {
    CloseableHttpAsyncClient client = HttpAsyncClientBuilder.create()
            .setDefaultRequestConfig(RequestConfig.custom().setConnectTimeout(1234).build())
            .build();
    HttpComponentsAsyncClientHttpRequestFactory factory = new HttpComponentsAsyncClientHttpRequestFactory(client);

    URI uri = new URI(baseUrl + "/status/ok");
    HttpComponentsAsyncClientHttpRequest request = (HttpComponentsAsyncClientHttpRequest)
            factory.createAsyncRequest(uri, HttpMethod.GET);

    assertNull("No custom config should be set with a custom HttpClient",
            request.getHttpContext().getAttribute(HttpClientContext.REQUEST_CONFIG));

    factory.setConnectionRequestTimeout(4567);
    HttpComponentsAsyncClientHttpRequest request2 = (HttpComponentsAsyncClientHttpRequest)
            factory.createAsyncRequest(uri, HttpMethod.GET);
    Object requestConfigAttribute = request2.getHttpContext().getAttribute(HttpClientContext.REQUEST_CONFIG);
    assertNotNull(requestConfigAttribute);
    RequestConfig requestConfig = (RequestConfig) requestConfigAttribute;

    assertEquals(4567, requestConfig.getConnectionRequestTimeout());
    // No way to access the request config of the HTTP client so no way to "merge" our customizations
    assertEquals(-1, requestConfig.getConnectTimeout());
}

/**
 * Creates an asynchronous HTTP client configuration with default timeouts.
 *
 * @see #newHttpAsyncClient(boolean)
 */
protected static CloseableHttpAsyncClient newHttpAsyncClient(boolean useSSL) {
    RequestConfig requestConfig = RequestConfig.custom().setSocketTimeout(DEFAULT_SOCKET_TIMEOUT)
            .setConnectTimeout(DEFAULT_CONNECT_TIMEOUT).build();

    HttpAsyncClientBuilder builder = HttpAsyncClients.custom();

    if (useSSL) {
        try {
            SSLContext context = SSLContext.getInstance("SSL");
            context.init(null, new TrustManager[]{new TrustAllX509Manager()}, new SecureRandom());

            SSLIOSessionStrategy strategy = new SSLIOSessionStrategy(context,
                    SSLIOSessionStrategy.getDefaultHostnameVerifier());

            builder.setSSLStrategy(strategy);
        } catch (Exception e) {
            log.error("Failed initializing SSL context! Skipped.", e);
        }
    }

    return builder.setDefaultRequestConfig(requestConfig).build();
}

private CloseableHttpAsyncClient getAsyncProxyHttpClient(URI proxyUri) {
  LOG.info("Creating async proxy http client");
  PoolingNHttpClientConnectionManager cm = getAsyncConnectionManager();
  HttpHost proxy = new HttpHost(proxyUri.getHost(), proxyUri.getPort());

  HttpAsyncClientBuilder clientBuilder = HttpAsyncClients.custom();
  if (cm != null) {
    clientBuilder = clientBuilder.setConnectionManager(cm);
  }

  if (proxy != null) {
    clientBuilder = clientBuilder.setProxy(proxy);
  }
  clientBuilder = setRedirects(clientBuilder);
  return clientBuilder.build();
}

private HttpAsyncClientBuilder setRedirects(HttpAsyncClientBuilder clientBuilder) {
  clientBuilder.setRedirectStrategy(new DefaultRedirectStrategy() {
    /** Redirectable methods. */
    private String[] REDIRECT_METHODS = new String[] { 
      HttpGet.METHOD_NAME, HttpPost.METHOD_NAME, 
      HttpPut.METHOD_NAME, HttpDelete.METHOD_NAME, HttpHead.METHOD_NAME 
    };

    @Override
    protected boolean isRedirectable(String method) {
      for (String m : REDIRECT_METHODS) {
        if (m.equalsIgnoreCase(method)) {
          return true;
        }
      }
      return false;
    }
  });
  return clientBuilder;
}

private QueryExecutor(String userAgent, ObjectMapper mapper, HttpHost proxy)
{
    checkNotNull(userAgent, "userAgent is null");
    checkNotNull(mapper, "mapper is null");

    this.userAgent = userAgent;
    this.mapper = mapper;

    HttpClientBuilder builder = HttpClients.custom();
    HttpAsyncClientBuilder asyncBuilder = HttpAsyncClients.custom();

    if (proxy != null) {
        DefaultProxyRoutePlanner routePlanner = new DefaultProxyRoutePlanner(proxy);
        builder.setRoutePlanner(routePlanner);
        asyncBuilder.setRoutePlanner(routePlanner);
    }

    this.httpClient = asyncBuilder.build();
    this.httpClient.start();
}

/**
 * Used internally to initialize the internal HTTP client used by all
 * instances of a client.
 * <p>
 * This method can be overriden to provide a client with different options.
 * The client built gets an extra interceptor to add the credentials headers.
 *
 * @return HTTP default async client builder
 */
protected static HttpAsyncClientBuilder defaultClientBuilder() {

  try {
    DefaultConnectingIOReactor ioReactor = new DefaultConnectingIOReactor();
    connMgr = new PoolingNHttpClientConnectionManager(ioReactor);
    connMgr.setMaxTotal(maxConnections);
    connMgr.setDefaultMaxPerRoute(maxConnections);
  } catch (IOReactorException e) {
  }

  return HttpAsyncClients
      .custom()
      .addInterceptorLast(new GzipInterceptors.GzipRequestInterceptor())
      .setConnectionManager(connMgr)
      .setDefaultRequestConfig(
          RequestConfig.custom()
          .setSocketTimeout(3600 * 1000) // 1 hour
              .build())
      .setKeepAliveStrategy(keepAliveStrategy);
}

public HttpClient(Auth auth, String apiAddress, int maxConnections) {
    this.auth = auth;
    if(apiAddress != null) {
        this.apiAddress = apiAddress;
    } else {
        this.apiAddress = AlgorithmiaConf.apiAddress();
    }

    client = HttpAsyncClientBuilder.create()
        .setMaxConnTotal(maxConnections)
        .setMaxConnPerRoute(maxConnections)
        .useSystemProperties()
        .build();

    synchronized (clients) {
        clients.add(client);
    }
    client.start();
}

/**
 * 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;
}

public static RestClient createClient(ElasticsearchDatastoreProperties datastore) throws MalformedURLException {
    String urlStr = datastore.nodes.getValue();
    String[] urls = urlStr.split(",");
    HttpHost[] hosts = new HttpHost[urls.length];
    int i = 0;
    for (String address : urls) {
        URL url = new URL("http://" + address);
        hosts[i] = new HttpHost(url.getHost(), url.getPort(), url.getProtocol());
        i++;
    }
    RestClientBuilder restClientBuilder = RestClient.builder(hosts);
    if (datastore.auth.useAuth.getValue()) {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials(datastore.auth.userId.getValue(), datastore.auth.password.getValue()));
        restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {

            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
                return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            }
        });
    }
    return restClientBuilder.build();
}

@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 ApacheHttpAsyncClient(HttpAsyncClientBuilder builder, Config config, SharedResourcesBroker<GobblinScopeTypes> broker) {
  super (broker, HttpUtils.createApacheHttpClientLimiterKey(config));
  config = config.withFallback(FALLBACK);

  RequestConfig requestConfig = RequestConfig.copy(RequestConfig.DEFAULT)
      .setSocketTimeout(config.getInt(REQUEST_TIME_OUT_MS_KEY))
      .setConnectTimeout(config.getInt(CONNECTION_TIME_OUT_MS_KEY))
      .setConnectionRequestTimeout(config.getInt(CONNECTION_TIME_OUT_MS_KEY))
      .build();

  try {
    builder.disableCookieManagement().useSystemProperties().setDefaultRequestConfig(requestConfig);
    builder.setConnectionManager(getNHttpConnManager(config));
    client = builder.build();
    client.start();
  } catch (IOException e) {
    throw new RuntimeException("ApacheHttpAsyncClient cannot be initialized");
  }
}

private void startRestClient() {
    try {
        HttpHost[] hosts = new HttpHost[] {
                // new HttpHost("10.132.161.173", 30002, "http")
                new HttpHost("127.0.0.1", 9200, "http") };
        client = RestClient.builder(hosts).setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
            @Override
            public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                return requestConfigBuilder.setConnectTimeout(2000).setSocketTimeout(10000);
            }
        }).setMaxRetryTimeoutMillis(10000).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                return httpClientBuilder.setMaxConnPerRoute(100).setMaxConnTotal(200);
                // return httpClientBuilder
                // .setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(1).build());
            }
        }).build();
    } catch (Exception e) {
        LOG.error(e.getMessage(), e.fillInStackTrace());
    }
}

@Before
public void before() throws Exception {
  tsdb = PowerMockito.mock(TSDB.class);
  config = new Config(false);
  connection_manager = mock(PoolingNHttpClientConnectionManager.class);
  client_builder = mock(HttpAsyncClientBuilder.class);
  client = mock(CloseableHttpAsyncClient.class);
  ts_meta_schema = mock(TSMetaSchema.class);
  uid_meta_schema = mock(UIDMetaSchema.class);
  annotation_schema = mock(AnnotationSchema.class);

  config.overrideConfig("tsd.search.elasticsearch.host", "localhost:9200");

  when(tsdb.getConfig()).thenReturn(config);

  PowerMockito.mockStatic(HttpAsyncClients.class);
  when(HttpAsyncClients.custom()).thenReturn(client_builder);

  PowerMockito.whenNew(PoolingNHttpClientConnectionManager.class)
    .withAnyArguments().thenReturn(connection_manager);
  when(client_builder.build()).thenReturn(client);
}

private CloseableHttpAsyncClient createHttpClient() {

        final RequestConfig.Builder requestConfigBuilder = RequestConfig.custom();
        requestConfigBuilder.setConnectionRequestTimeout(TIMEOUT_CONNECTION_REQUEST);
        requestConfigBuilder.setConnectTimeout(TIMEOUT_CONNECT);
        requestConfigBuilder.setSocketTimeout(TIMEOUT_SOCKET);

        if (this.mProxyHost != null) {
            requestConfigBuilder.setProxy(this.mProxyHost);
        }

        final HttpAsyncClientBuilder clientBuilder = HttpAsyncClients.custom();

        final RequestConfig requestConfig = requestConfigBuilder.build();
        clientBuilder.setDefaultRequestConfig(requestConfig);

        if (this.mProxyHost != null && !StringUtils.isEmpty(this.mProxyUsername)) {
            final AuthScope authScope = new AuthScope(this.mProxyHost);
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            final Credentials credentials = new UsernamePasswordCredentials(this.mProxyUsername, this.mProxyPassword);
            credentialsProvider.setCredentials(authScope, credentials);
            clientBuilder.setDefaultCredentialsProvider(credentialsProvider);
        }

        return clientBuilder.build();
    }

public static CloseableHttpAsyncClient create(RequestConfig requestConfig) {
    HttpAsyncClientBuilder builder = HttpAsyncClients.custom();
    builder.setConnectionManager(connManager)
            .setDefaultCookieStore(cookieStore)
            .setDefaultCredentialsProvider(credentialsProvider);

    if (null != requestConfig) {
        return builder.setDefaultRequestConfig(requestConfig).build();
    } else {
        return builder.setDefaultRequestConfig(defaultRequestConfig)
                .build();
    }
}

private static RestClient createRestClient(final boolean useAuth, final boolean usePreemptiveAuth) {
    // provide the username/password for every request
    final BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider();
    credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("user", "pass"));

    final RestClientBuilder restClientBuilder = RestClient.builder(
        new HttpHost(httpServer.getAddress().getHostString(), httpServer.getAddress().getPort())).setDefaultHeaders(defaultHeaders);
    if (pathPrefix.length() > 0) {
        // sometimes cut off the leading slash
        restClientBuilder.setPathPrefix(randomBoolean() ? pathPrefix.substring(1) : pathPrefix);
    }

    if (useAuth) {
        restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(final HttpAsyncClientBuilder httpClientBuilder) {
                if (usePreemptiveAuth == false) {
                    // disable preemptive auth by ignoring any authcache
                    httpClientBuilder.disableAuthCaching();
                }

                return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            }
        });
    }

    return restClientBuilder.build();
}

public static CloseableHttpAsyncClient createSSLClient(HttpHost proxy) {
    TrustStrategy acceptingTrustStrategy = new TrustStrategy() {

        @Override
        public boolean isTrusted(X509Certificate[] arg0, String arg1) throws CertificateException {
            return true;
        }
    };

    try {
        SSLContext sslContext = SSLContexts.custom()
                .loadTrustMaterial(null, acceptingTrustStrategy)
                .build();

        HttpAsyncClientBuilder client = HttpAsyncClients.custom()
                .setDefaultCookieStore(new BasicCookieStore())
                .setSSLContext(sslContext)
                .setSSLHostnameVerifier(SSLConnectionSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);

        if (proxy !=null) {
            client.setProxy(proxy);
        }

        return client.build();
    } catch (Exception e) {
        System.out.println("Could not create SSLContext");
        return null;
    }

}

private void addProxy(RESTPool pool, HttpAsyncClientBuilder builder) {
    if (pool.getProxy() == null) return;

    Proxy proxy = pool.getProxy();

    if (proxy.getUsername() != null) {
        CredentialsProvider provider = makeProxyCredentialsProvider(proxy);
        builder.setDefaultCredentialsProvider(provider);
    }

    HttpHost proxyHost = new HttpHost(proxy.getHostname(), proxy.getPort());
    builder.setRoutePlanner(new DefaultProxyRoutePlanner(proxyHost));
}

public static RestClient createRestClient(String url, int port, CredentialsProvider credentialsProvider) {
    RestClient restClient = RestClient.builder(new HttpHost(url, port))
            .setHttpClientConfigCallback((HttpAsyncClientBuilder httpClientBuilder)
                    -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)).build();

    return restClient;
}

/**
 *
 * @param url
 * @param port
 * @param credentialsProvider
 * @return
 */
@Override
public RestClient createRestClient(String url, int port, CredentialsProvider credentialsProvider) {
    RestClient restClient = RestClient.builder(new HttpHost(url, port))
            .setHttpClientConfigCallback((HttpAsyncClientBuilder httpClientBuilder)
                    -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)).build();

    return restClient;
}

private static synchronized GCCloseable<CloseableHttpAsyncClient> get_http_client_() {
    GCCloseable<CloseableHttpAsyncClient> result = http_client_.get();
    if (result != null && !result.get().isRunning())
        result = null;  // Reactor appears to spontaneously shut down.
    if (result == null) {
        result = new GCCloseable<>(HttpAsyncClientBuilder.create()
                .useSystemProperties()
                .build());
        result.get().start();
        http_client_ = new WeakReference<>(result);
    }
    return result;
}

protected CloseableHttpAsyncClient configuredHttpAsyncClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
    LOG.info("HttpAsyncClient settings: maxConnections: {}, socketTimeout: {}, connectTimeout: {}",
            maxConnections, socketTimeout, connectTimeout);
    return httpAsyncClientBuilder.setMaxConnPerRoute(maxConnections)
            .setMaxConnTotal(maxConnections)
            .setDefaultRequestConfig(RequestConfig.custom()
                    .setSocketTimeout(socketTimeout)
                    .setConnectTimeout(connectTimeout)
                    .build())
            .disableCookieManagement()
            .build();
}

public ApiHttpClient(final String channelAccessToken) {

        RequestConfig requestConfig = RequestConfig.custom()
                .setConnectionRequestTimeout(timeoutInMillis)
                .setConnectTimeout(timeoutInMillis)
                .build();

        CloseableHttpAsyncClient asyncClient = HttpAsyncClientBuilder.create()
                .setDefaultRequestConfig(requestConfig)
                .addInterceptorLast((HttpRequest httpRequest, HttpContext httpContext) -> {
                    httpRequest.addHeader("X-Line-ChannelToken", channelAccessToken);
                    httpRequest.addHeader("Content-Type", "application/json; charser=UTF-8");
                    httpRequest.removeHeaders("Accept");
                    httpRequest.addHeader("Accept", "application/json; charset=UTF-8");
                })
                .setMaxConnTotal(maxConnections)
                .setMaxConnPerRoute(maxConnections)
                .disableCookieManagement()
                .build();

        asyncRestTemplate = new AsyncRestTemplate(new HttpComponentsAsyncClientHttpRequestFactory(asyncClient));
        asyncRestTemplate.setErrorHandler(new ApiResponseErrorHandler());

        httpHeaders = new HttpHeaders();
        httpHeaders.set("X-Line-ChannelToken", channelAccessToken);
        httpHeaders.setContentType(new MediaType("application", "json", Charset.forName("UTF-8")));
        List<MediaType> list = new ArrayList<>();
        list.add(new MediaType("application", "json", Charset.forName("UTF-8")));
        httpHeaders.setAccept(list);

        objectMapper = new ObjectMapper();
        objectMapper.configure(MapperFeature.USE_WRAPPER_NAME_AS_PROPERTY_NAME, true);
        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        objectMapper.configure(SerializationFeature.WRITE_NULL_MAP_VALUES, false);
        objectMapper.setAnnotationIntrospector(new JaxbAnnotationIntrospector(TypeFactory.defaultInstance()));
        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
    }

public CloseableHttpAsyncClient createHttpAsyncClient(
        ClientConfiguration config, PoolingNHttpClientConnectionManager cm) {
    HttpAsyncClientBuilder httpClientBuilder = HttpAsyncClients.custom();
    httpClientBuilder.setConnectionManager(cm);
    RequestConfig requestConfig = RequestConfig.custom()
            .setConnectTimeout(config.getConnectionTimeoutInMillisecond())
            .setSocketTimeout(config.getSocketTimeoutInMillisecond()).build();
    httpClientBuilder.setDefaultRequestConfig(requestConfig);
    httpClientBuilder.setUserAgent(config.getUserAgent());
    httpClientBuilder.disableCookieManagement();
    String proxyHost = config.getProxyHost();
    int proxyPort = config.getProxyPort();
    if (proxyHost != null) {
        if (proxyPort <= 0) {
            throw new ClientException("The proxy port is invalid. Please check your configuration.");
        }
        HttpHost proxy = new HttpHost(proxyHost, proxyPort);
        httpClientBuilder.setProxy(proxy);
        String proxyUsername = config.getProxyUsername();
        String proxyPassword = config.getProxyPassword();
        if (proxyUsername != null && proxyPassword != null) {
            String proxyDomain = config.getProxyDomain();
            String proxyWorkstation = config.getProxyWorkstation();
            CredentialsProvider credsProvider = new BasicCredentialsProvider();
            credsProvider.setCredentials(
                    new AuthScope(proxyHost, proxyPort), new NTCredentials(
                            proxyUsername, proxyPassword, proxyWorkstation,
                            proxyDomain));
            httpClientBuilder.setDefaultCredentialsProvider(credsProvider);
        }
    }
    return httpClientBuilder.build();
}

@Bean
public CloseableHttpAsyncClient asyncHttpClient(PoolingNHttpClientConnectionManager poolingNHttpClientConnectionManager) {

    RequestConfig config = RequestConfig.custom()
            .setConnectTimeout(requestTimeout)
            .setSocketTimeout(requestTimeout)
            .setConnectionRequestTimeout(requestTimeout)
            .build();

    return HttpAsyncClientBuilder
            .create().setConnectionManager(poolingNHttpClientConnectionManager)
            .setDefaultRequestConfig(config).build();
}

/**
 * Create Olingo2 Application with custom HTTP client builder.
 *
 * @param serviceUri Service Application base URI.
 * @param builder custom HTTP client builder.
 */
public Olingo2AppImpl(String serviceUri, HttpAsyncClientBuilder builder) {
    setServiceUri(serviceUri);

    if (builder == null) {
        this.client = HttpAsyncClients.createDefault();
    } else {
        this.client = builder.build();
    }
    this.client.start();
    this.contentType = ContentType.create("application/json", Consts.UTF_8);
}

/**
 * @param analyticsId The Google Analytics ID to which reports will be sent.
 * @param clientBuilder A builder for HTTP client objects. This injection is for testing; most
 *        clients should never need to bind anything but the default implementation here.
 * @param random A random number generator for cache-busting. This injection is for testing; most
 *        clients should never need to bind anything but the default implementation here.
 */
@Inject
public AsyncMetricsSender(@GoogleAnalyticsId String analyticsId,
    HttpAsyncClientBuilder clientBuilder, Random random) {
  this.analyticsId = analyticsId;
  this.random = random;
  this.client = clientBuilder.setUserAgent(MetricsUtils.USER_AGENT).build();
  this.client.start();
}

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

}

@Override
public void open() {
    if (open) {
        // Ignore
        return;
    }
    final HttpAsyncClientBuilder builder = HttpAsyncClients.custom()
            .setThreadFactory(threadFactory);
    if (credentialsProvider != null) {
        builder.setDefaultCredentialsProvider(credentialsProvider);
    }
    httpclient = builder.build();
    httpclient.start();
    this.open = true;
}

private static void startRestClient() throws IOException {
    if (client == null) {
        client = RestClient.builder(new HttpHost(testClusterHost, testClusterRestPort, testClusterScheme)).build();

        securityInstalled = testClusterRunning(false);
        if (securityInstalled) {
            // We have a secured cluster. So we need to create a secured client
            // But first we need to close the previous client we built
            if (client != null) {
                client.close();
            }

            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY,
                    new UsernamePasswordCredentials(testClusterUser, testClusterPass));

            client = RestClient.builder(new HttpHost(testClusterHost, testClusterRestPort, testClusterScheme))
                    .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                        @Override
                        public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                            return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                        }
                    })
                    .build();
            securityInstalled = testClusterRunning(true);
        }
    }
}

/**
 * Creates default proxy route planner.
 *
 * @param settings
 *         settings to use.
 * @param httpClientBuilder
 *         http client builder.
 * @return Instance of {@link DefaultProxyRoutePlanner}.
 */
private DefaultProxyRoutePlanner createProxyRoutePlanner(HttpSettings settings, HttpAsyncClientBuilder httpClientBuilder) {
    HttpHost proxy = new HttpHost(settings.getProxyUrl().getHost(), settings.getProxyUrl().getPort());
    if (settings.getProxyUser() != null) {
        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        String proxyUser = settings.getProxyUser();
        String proxyPassword = settings.getProxyPassword();
        UsernamePasswordCredentials credentials = new UsernamePasswordCredentials(proxyUser, proxyPassword);
        credentialsProvider.setCredentials(new AuthScope(proxy), credentials);
        httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
    }
    return new DefaultProxyRoutePlanner(proxy);
}