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

@Override
public CloseableHttpAsyncClient generateClient ()
{
   CredentialsProvider credsProvider = new BasicCredentialsProvider();
   credsProvider.setCredentials(new AuthScope (AuthScope.ANY),
           new UsernamePasswordCredentials(serviceUser, servicePass));
   RequestConfig rqconf = RequestConfig.custom()
         .setCookieSpec(CookieSpecs.DEFAULT)
         .setSocketTimeout(Timeouts.SOCKET_TIMEOUT)
         .setConnectTimeout(Timeouts.CONNECTION_TIMEOUT)
         .setConnectionRequestTimeout(Timeouts.CONNECTION_REQUEST_TIMEOUT)
         .build();
   CloseableHttpAsyncClient res = HttpAsyncClients.custom ()
         .setDefaultCredentialsProvider (credsProvider)
         .setDefaultRequestConfig(rqconf)
         .build ();
   res.start ();
   return res;
}

public void send(List<Map<Object, Object>> events, AsyncSuccessCallback<ProducedEventsResult> onSuccess, AsyncFailCallback onFail, AsyncCancelledCallback onCancel) throws IOException, InterruptedException {
    CloseableHttpAsyncClient httpClient = HttpAsyncClients.createDefault();
    httpClient.start();

    String url = String.format("%s/%s/bulk-produce", this.endpoint, this.topicId);
    System.out.println(url);
    HttpPost httpPost = new HttpPost(url);
    httpPost.addHeader("Authorization", this.writeApiKey);
    httpPost.addHeader("Content-type", this.format);

    String jsonString = MAPPER.writeValueAsString(events);
    HttpEntity entity = new ByteArrayEntity(jsonString.getBytes());
    httpPost.setEntity(entity);

    ResponseParser<ProducedEventsResult> parser = new BulkProduceEventsParser();
    AsyncCallback cb = new AsyncCallback(httpClient, parser, MAPPER, onSuccess, onFail, onCancel);
    httpClient.execute(httpPost, cb);
}

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

@SuppressWarnings("unchecked")
@Before
public void before() throws Exception {
  config = new ESPluginConfig(new Config(false));
  client = mock(CloseableHttpAsyncClient.class);
  es = mock(ElasticSearch.class);
  meta = new UIDMeta(UniqueIdType.METRIC, new byte[] { 1 }, "sys.cpu.user");
  index = config.getString("tsd.search.elasticsearch.index");
  doc_type = config.getString("tsd.search.elasticsearch.uidmeta_type");

  when(es.httpClient()).thenReturn(client);
  when(es.host()).thenReturn(HOST);
  when(es.index()).thenReturn(index);
  when(es.config()).thenReturn(config);
  when(client.execute(any(HttpUriRequest.class), 
      any(FutureCallback.class)))
    .thenAnswer(new Answer<Void>() {
      @Override
      public Void answer(InvocationOnMock invocation) throws Throwable {
        request = (HttpUriRequest) invocation.getArguments()[0];
        cb = (FutureCallback<HttpResponse>) invocation.getArguments()[1];
        return null;
      }
    });
}

@SuppressWarnings({ "unchecked", "rawtypes" })
public void testTooLargeResponse() throws Exception {
    ContentTooLongException tooLong = new ContentTooLongException("too long!");
    CloseableHttpAsyncClient httpClient = mock(CloseableHttpAsyncClient.class);
    when(httpClient.<HttpResponse>execute(any(HttpAsyncRequestProducer.class), any(HttpAsyncResponseConsumer.class),
            any(HttpClientContext.class), any(FutureCallback.class))).then(new Answer<Future<HttpResponse>>() {
        @Override
        public Future<HttpResponse> answer(InvocationOnMock invocationOnMock) throws Throwable {
            HeapBufferedAsyncResponseConsumer consumer = (HeapBufferedAsyncResponseConsumer) invocationOnMock.getArguments()[1];
            FutureCallback callback = (FutureCallback) invocationOnMock.getArguments()[3];
            assertEquals(new ByteSizeValue(100, ByteSizeUnit.MB).bytesAsInt(), consumer.getBufferLimit());
            callback.failed(tooLong);
            return null;
        }
    });
    RemoteScrollableHitSource source = sourceWithMockedClient(true, httpClient);

    AtomicBoolean called = new AtomicBoolean();
    Consumer<Response> checkResponse = r -> called.set(true);
    Throwable e = expectThrows(RuntimeException.class,
            () -> source.doStartNextScroll(FAKE_SCROLL_ID, timeValueMillis(0), checkResponse));
    // Unwrap the some artifacts from the test
    while (e.getMessage().equals("failed")) {
        e = e.getCause();
    }
    // This next exception is what the user sees
    assertEquals("Remote responded with a chunk that was too large. Use a smaller batch size.", e.getMessage());
    // And that exception is reported as being caused by the underlying exception returned by the client
    assertSame(tooLong, e.getCause());
    assertFalse(called.get());
}

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

@Before
@SuppressWarnings("unchecked")
public void createRestClient() throws IOException {
    CloseableHttpAsyncClient httpClient = mock(CloseableHttpAsyncClient.class);
    when(httpClient.<HttpResponse>execute(any(HttpAsyncRequestProducer.class), any(HttpAsyncResponseConsumer.class),
           any(HttpClientContext.class), any(FutureCallback.class))).thenAnswer(new Answer<Future<HttpResponse>>() {
        @Override
        public Future<HttpResponse> answer(InvocationOnMock invocationOnMock) throws Throwable {
            HttpAsyncRequestProducer requestProducer = (HttpAsyncRequestProducer) invocationOnMock.getArguments()[0];
            HttpUriRequest request = (HttpUriRequest)requestProducer.generateRequest();
            HttpHost httpHost = requestProducer.getTarget();
            HttpClientContext context = (HttpClientContext) invocationOnMock.getArguments()[2];
            assertThat(context.getAuthCache().get(httpHost), instanceOf(BasicScheme.class));
            FutureCallback<HttpResponse> futureCallback = (FutureCallback<HttpResponse>) invocationOnMock.getArguments()[3];
            //return the desired status code or exception depending on the path
            if (request.getURI().getPath().equals("/soe")) {
                futureCallback.failed(new SocketTimeoutException(httpHost.toString()));
            } else if (request.getURI().getPath().equals("/coe")) {
                futureCallback.failed(new ConnectTimeoutException(httpHost.toString()));
            } else if (request.getURI().getPath().equals("/ioe")) {
                futureCallback.failed(new IOException(httpHost.toString()));
            } else {
                int statusCode = Integer.parseInt(request.getURI().getPath().substring(1));
                StatusLine statusLine = new BasicStatusLine(new ProtocolVersion("http", 1, 1), statusCode, "");
                futureCallback.completed(new BasicHttpResponse(statusLine));
            }
            return null;
        }
    });
    int numHosts = RandomNumbers.randomIntBetween(getRandom(), 2, 5);
    httpHosts = new HttpHost[numHosts];
    for (int i = 0; i < numHosts; i++) {
        httpHosts[i] = new HttpHost("localhost", 9200 + i);
    }
    failureListener = new HostsTrackingFailureListener();
    restClient = new RestClient(httpClient, 10000, new Header[0], httpHosts, null, failureListener);
}

@Override
public CloseableHttpAsyncClient generateClient ()
{
   CredentialsProvider credsProvider = new BasicCredentialsProvider();
   credsProvider.setCredentials(new AuthScope (AuthScope.ANY),
            new UsernamePasswordCredentials(username, password));
   RequestConfig rqconf = RequestConfig.custom()
         .setCookieSpec(CookieSpecs.DEFAULT)
         .setSocketTimeout(Timeouts.SOCKET_TIMEOUT)
         .setConnectTimeout(Timeouts.CONNECTION_TIMEOUT)
         .setConnectionRequestTimeout(Timeouts.CONNECTION_REQUEST_TIMEOUT)
         .build();
   CloseableHttpAsyncClient res = HttpAsyncClients.custom ()
         .setDefaultCredentialsProvider (credsProvider)
         .setDefaultRequestConfig(rqconf)
         .build ();
   res.start ();
   return res;
}

public static HttpClient createHttpClient(HiTSDBConfig config) throws HttpClientInitException {
    Objects.requireNonNull(config);

    // 创建 ConnectingIOReactor
    ConnectingIOReactor ioReactor = initIOReactorConfig(config);

    // 创建链接管理器
    final PoolingNHttpClientConnectionManager cm = new PoolingNHttpClientConnectionManager(ioReactor);

    // 创建令牌管理器
    semaphoreManager = createSemaphoreManager(config);

    // 创建HttpAsyncClient
    CloseableHttpAsyncClient httpAsyncClient = createPoolingHttpClient(config,cm,semaphoreManager);

    // 组合生产HttpClientImpl
    HttpClient httpClientImpl = new HttpClient(config,httpAsyncClient,semaphoreManager);

    return httpClientImpl;
}

public void send(Map<Object, Object> event, AsyncSuccessCallback<ProducedEventResult> onSuccess, AsyncFailCallback onFail, AsyncCancelledCallback onCancel) throws IOException {
    CloseableHttpAsyncClient httpClient = HttpAsyncClients.createDefault();
    httpClient.start();

    String url = String.format("%s/%s/produce", this.endpoint, this.topicId);
    HttpPost httpPost = new HttpPost(url);
    httpPost.addHeader("Authorization", this.writeApiKey);
    httpPost.addHeader("Content-type", this.format);

    String jsonString = MAPPER.writeValueAsString(event);
    HttpEntity entity = new ByteArrayEntity(jsonString.getBytes());
    httpPost.setEntity(entity);

    ResponseParser<ProducedEventResult> parser = new ProduceEventParser();
    AsyncCallback cb = new AsyncCallback(httpClient, parser, MAPPER, onSuccess, onFail, onCancel);
    httpClient.execute(httpPost, cb);
}

RestClient(String baseUrl,
           ObjectMapper objectMapper,
           Map<String, Object> defaultHeaders,
           Function<String, String> urlTransformer,
           PoolingNHttpClientConnectionManager asyncConnectionManager,
           PoolingHttpClientConnectionManager syncConnectionManager,
           CloseableHttpAsyncClient asyncClient,
           CloseableHttpClient syncClient) {
    this.objectMapper = objectMapper;
    this.baseUrl = baseUrl;
    this.urlTransformer = urlTransformer;
    this.asyncConnectionManager = asyncConnectionManager;
    this.syncConnectionManager = syncConnectionManager;
    this.asyncClient = asyncClient;
    this.syncClient = syncClient;
    this.defaultHeaders.putAll(defaultHeaders);
    this.id = UUID.randomUUID().toString().substring(0, 8);
}

public static void main(String[] args) throws Exception {
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
    try {
        httpclient.start();
        HttpHost proxy = new HttpHost("someproxy", 8080);
        RequestConfig config = RequestConfig.custom().setProxy(proxy).build();
        HttpGet request = new HttpGet("https://issues.apache.org/");
        request.setConfig(config);
        Future<HttpResponse> future = httpclient.execute(request, null);
        HttpResponse response = future.get();
        System.out.println("Response: " + response.getStatusLine());
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
}

public static void main(final String[] args) throws Exception {
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
    try {
        httpclient.start();
        Future<Boolean> future = httpclient.execute(HttpAsyncMethods.createGet("http://localhost:8080/"),
                new MyResponseConsumer(), null);
        Boolean result = future.get();
        if (result != null && result.booleanValue()) {
            System.out.println("Request successfully executed");
        } else {
            System.out.println("Request failed");
        }
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
    System.out.println("Done");
}

public static void main(String[] args) throws Exception {
    CredentialsProvider credsProvider = new BasicCredentialsProvider();
    credsProvider.setCredentials(new AuthScope("localhost", 443),
            new UsernamePasswordCredentials("username", "password"));
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setDefaultCredentialsProvider(credsProvider)
            .build();
    try {
        HttpGet httpget = new HttpGet("http://localhost/");

        System.out.println("Executing request " + httpget.getRequestLine());
        Future<HttpResponse> future = httpclient.execute(httpget, null);
        HttpResponse response = future.get();
        System.out.println("Response: " + response.getStatusLine());
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
}

public static void main(String[] args) throws Exception {
    CredentialsProvider credsProvider = new BasicCredentialsProvider();
    credsProvider.setCredentials(new AuthScope("someproxy", 8080),
            new UsernamePasswordCredentials("username", "password"));
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setDefaultCredentialsProvider(credsProvider)
            .build();
    try {
        httpclient.start();
        HttpHost proxy = new HttpHost("someproxy", 8080);
        RequestConfig config = RequestConfig.custom().setProxy(proxy).build();
        HttpGet httpget = new HttpGet("https://issues.apache.org/");
        httpget.setConfig(config);
        Future<HttpResponse> future = httpclient.execute(httpget, null);
        HttpResponse response = future.get();
        System.out.println("Response: " + response.getStatusLine());
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
}

@Override
public ResponseEntity<String> submit(JCurlRequestOptions requestOptions) throws Exception {
    ResponseEntity<String> stringResponseEntity = null;
    try (CloseableHttpAsyncClient hc = createCloseableHttpAsyncClient()) {
        for (int i = 0; i < requestOptions.getCount(); i++) {
            final HttpHeaders headers = new HttpHeaders();
            for (Map.Entry<String, String> e : requestOptions.getHeaderMap().entrySet()) {
                headers.put(e.getKey(), Collections.singletonList(e.getValue()));
            }

            final HttpEntity<Void> requestEntity = new HttpEntity<>(headers);

            AsyncRestTemplate template = new AsyncRestTemplate(new HttpComponentsAsyncClientHttpRequestFactory(hc));
            final ListenableFuture<ResponseEntity<String>> exchange = template.exchange(requestOptions.getUrl(), HttpMethod.GET, requestEntity, String.class);
            stringResponseEntity = exchange.get();
            System.out.println(stringResponseEntity.getBody());

        }
        return stringResponseEntity;
    }
}

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

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

public final static HttpResponse get(String url, List<NameValuePair> parameters) throws Throwable {
    CloseableHttpAsyncClient client = HttpAsyncClients.createDefault();

    StringBuffer sb = new StringBuffer("?");
    for (NameValuePair pair : parameters) {
        sb.append(pair.getName());
        sb.append("=");
        sb.append(URLEncoder.encode(pair.getValue(), "UTF-8"));
        sb.append("&");
    }

    client.start();
    final HttpGet httpGet = new HttpGet(url + sb.toString());
    httpGet.removeHeaders("X-FORWARDED-FOR");
    httpGet.setHeader("X-FORWARDED-FOR", Environment.LOCAL_IP_ADDR);

    logger.debug("-> GET " + (url + sb.toString()));

    Future<HttpResponse> future = client.execute(httpGet, null);
    HttpResponse resp = future.get();

    return resp;
}

public final static HttpResponse post(String url, List<NameValuePair> parameters) throws Throwable {
    CloseableHttpAsyncClient client = HttpAsyncClients.createDefault();

    client.start();
    final HttpPost httpPost = new HttpPost(url);
    httpPost.removeHeaders("X-FORWARDED-FOR");
    httpPost.setHeader("X-FORWARDED-FOR", Environment.LOCAL_IP_ADDR);
    UrlEncodedFormEntity encodedFormEntity = new UrlEncodedFormEntity(parameters, "UTF-8");
    httpPost.setEntity(encodedFormEntity);

    logger.debug("-> POST " + url + " Parameters " + JsonUtil.toString(parameters));

    Future<HttpResponse> future = client.execute(httpPost, null);
    HttpResponse resp = future.get();

    return resp;
}

public final static HttpResponse post(String url, Object model) throws Throwable {
    CloseableHttpAsyncClient client = HttpAsyncClients.createDefault();

    client.start();
    final HttpPost httpPost = new HttpPost(url);
    httpPost.removeHeaders("X-FORWARDED-FOR");
    httpPost.setHeader("X-FORWARDED-FOR", Environment.LOCAL_IP_ADDR);
    StringEntity stringEntity = new StringEntity(JsonUtil.toString(model), ENCODING);
    stringEntity.setContentType(CONTENT_TYPE);
    httpPost.setEntity(stringEntity);

    logger.debug("-> POST " + url + " Parameters " + JsonUtil.toString(model));

    Future<HttpResponse> future = client.execute(httpPost, null);
    HttpResponse resp = future.get();

    return resp;
}

public final static HttpResponse post(String url, Object model, HttpEntity httpEntity) throws Throwable {
    CloseableHttpAsyncClient client = HttpAsyncClients.createDefault();

    client.start();
    final HttpPost httpPost = new HttpPost(url);
    httpPost.removeHeaders("X-FORWARDED-FOR");
    httpPost.setHeader("X-FORWARDED-FOR", Environment.LOCAL_IP_ADDR);
    httpPost.setEntity(httpEntity);

    logger.debug("-> POST " + url + " httpEntity  Parameters " + JsonUtil.toString(model));

    Future<HttpResponse> future = client.execute(httpPost, null);
    HttpResponse resp = future.get();

    return resp;
}

@SuppressWarnings("unchecked")
public void begin() throws InterruptedException {
    CloseableHttpAsyncClient httpclient = httpAsyncClientBuilder.build();
    httpclient.start();
    new Thread(() -> {
        while (true) {
            try {
                Url url = this.urlQueue.take();
                httpclient.execute(HttpAsyncMethods.createGet(url.url), new MyResponseConsumer(url), new MyFutureCallback(url));
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }).start();

}

private CompletableFuture<Stream<Metric>> do_request_(GroupName args, String url) {
    /*
     * Client seems to spontaneously stop its reactor.
     * No idea why, so in order to keep the reactor shutdown from disabling
     * the monitor entirely, we allow for resetting it.
     *
     * Since restarted clients drop their reference, keep a reference to the client locally.
     * The client needs to stay alive until the request completes,
     * so we attach it to the response consumer as a keep_live object.
     */
    final GCCloseable<CloseableHttpAsyncClient> client = httpClient();

    final CompletableFuture<Stream<Metric>> result = new CompletableFuture<>();  // Filled in by HttpResponseConsumer instance.
    final HttpGet request = new HttpGet(url);
    request.setConfig(request_config_);
    client.get().execute(request, new HttpResponseConsumer(result, args, url, client));
    return result;
}

/**
 * Initializes and returns the httpClient with NoopHostnameVerifier
 * 
 * @return CloseableHttpAsyncClient
 */
@Override
public CloseableHttpAsyncClient getHttpClient() {
    // Trust own CA and all self-signed certs
    SSLContext sslcontext = NonValidatingSSLSocketFactory.getSSLContext();
    // Allow TLSv1 protocol only

    SSLIOSessionStrategy sslSessionStrategy = new SSLIOSessionStrategy(sslcontext, new String[] { "TLSv1" }, null,
            new NoopHostnameVerifier());
    List<Header> headers = LogInsightClient.getDefaultHeaders();

    asyncHttpClient = HttpAsyncClients.custom().setSSLStrategy(sslSessionStrategy).setDefaultHeaders(headers)
            .build();
    asyncHttpClient.start();

    return asyncHttpClient;
}

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

/**
 * Constructs a RestClient object.
 *
 * @param target The address of the RESTful endpoint.
 * @param httpClient The asynchronous HTTP client.
 */
public RestClient(String target, CloseableHttpAsyncClient httpClient) {
  if (null == httpClient) {
    throw new IllegalArgumentException("Client cannot be null");
  }

  if (null == target) {
    throw new IllegalArgumentException("Target cannot be null");
  }

  this.target = target;
  this.httpClient = httpClient;

  if (!this.httpClient.isRunning()) {
    this.httpClient.start();
  }
}

private DashboardSetupStatus initDashboard(final String hostAddress, final int port) {
  final String dashboardURL = String.format("http://%s:%d/", hostAddress, port);
  try {
    // Create a pool of http client connection, which allow up to Integer.MAX_VALUE connections.
    final PoolingNHttpClientConnectionManager connectionManager
        = new PoolingNHttpClientConnectionManager(new DefaultConnectingIOReactor());
    connectionManager.setMaxTotal(Integer.MAX_VALUE);
    final CloseableHttpAsyncClient reusableHttpClient =
        HttpAsyncClients.custom().setConnectionManager(connectionManager).build();
    reusableHttpClient.start();

    // run another thread to send metrics.
    runMetricsSenderThread();

    return DashboardSetupStatus.getSuccessful(dashboardURL, reusableHttpClient);
  } catch (IOReactorException e) {
    LOG.log(Level.WARNING, "Dashboard: Fail on initializing connection to the dashboard server.", e);
    return DashboardSetupStatus.getFailed();
  }
}

@Override
public CloseableHttpAsyncClient generateClient ()
{
   CredentialsProvider credsProvider = new BasicCredentialsProvider();
   credsProvider.setCredentials(new AuthScope (AuthScope.ANY),
           new UsernamePasswordCredentials(serviceUser, servicePass));
   RequestConfig rqconf = RequestConfig.custom()
         .setCookieSpec(CookieSpecs.DEFAULT)
         .setSocketTimeout(Timeouts.SOCKET_TIMEOUT)
         .setConnectTimeout(Timeouts.CONNECTION_TIMEOUT)
         .setConnectionRequestTimeout(Timeouts.CONNECTION_REQUEST_TIMEOUT)
         .build();
   CloseableHttpAsyncClient res = HttpAsyncClients.custom ()
         .setDefaultCredentialsProvider (credsProvider)
         .setDefaultRequestConfig(rqconf)
         .build ();
   res.start ();
   return res;
}

@Override
public CloseableHttpAsyncClient generateClient ()
{
   CredentialsProvider credsProvider = new BasicCredentialsProvider();
   credsProvider.setCredentials(new AuthScope (AuthScope.ANY),
            new UsernamePasswordCredentials(username, password));
   RequestConfig rqconf = RequestConfig.custom()
         .setCookieSpec(CookieSpecs.DEFAULT)
         .setSocketTimeout(Timeouts.SOCKET_TIMEOUT)
         .setConnectTimeout(Timeouts.CONNECTION_TIMEOUT)
         .setConnectionRequestTimeout(Timeouts.CONNECTION_REQUEST_TIMEOUT)
         .build();
   CloseableHttpAsyncClient res = HttpAsyncClients.custom ()
         .setDefaultCredentialsProvider (credsProvider)
         .setDefaultRequestConfig(rqconf)
         .build ();
   res.start ();
   return res;
}

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

public static void main(String[] args)throws Exception {
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
    try {
        httpclient.start();
        HttpHost proxy = new HttpHost("someproxy", 8080);
        RequestConfig config = RequestConfig.custom()
                .setProxy(proxy)
                .build();
        HttpGet request = new HttpGet("https://issues.apache.org/");
        request.setConfig(config);
        Future<HttpResponse> future = httpclient.execute(request, null);
        HttpResponse response = future.get();
        System.out.println("Response: " + response.getStatusLine());
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
}

public static void main(final String[] args) throws Exception {
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
    try {
        httpclient.start();
        Future<Boolean> future = httpclient.execute(
                HttpAsyncMethods.createGet("http://localhost:8080/"),
                new MyResponseConsumer(), null);
        Boolean result = future.get();
        if (result != null && result.booleanValue()) {
            System.out.println("Request successfully executed");
        } else {
            System.out.println("Request failed");
        }
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
    System.out.println("Done");
}

public static void main(String[] args) throws Exception {
    CredentialsProvider credsProvider = new BasicCredentialsProvider();
    credsProvider.setCredentials(
            new AuthScope("localhost", 443),
            new UsernamePasswordCredentials("username", "password"));
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom()
            .setDefaultCredentialsProvider(credsProvider)
            .build();
    try {
        HttpGet httpget = new HttpGet("http://localhost/");

        System.out.println("Executing request " + httpget.getRequestLine());
        Future<HttpResponse> future = httpclient.execute(httpget, null);
        HttpResponse response = future.get();
        System.out.println("Response: " + response.getStatusLine());
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
}

public static void main(String[] args)throws Exception {
    CredentialsProvider credsProvider = new BasicCredentialsProvider();
    credsProvider.setCredentials(
            new AuthScope("someproxy", 8080),
            new UsernamePasswordCredentials("username", "password"));
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom()
            .setDefaultCredentialsProvider(credsProvider)
            .build();
    try {
        httpclient.start();
        HttpHost proxy = new HttpHost("someproxy", 8080);
        RequestConfig config = RequestConfig.custom()
                .setProxy(proxy)
                .build();
        HttpGet httpget = new HttpGet("https://issues.apache.org/");
        httpget.setConfig(config);
        Future<HttpResponse> future = httpclient.execute(httpget, null);
        HttpResponse response = future.get();
        System.out.println("Response: " + response.getStatusLine());
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
}

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