set all config objects on the async client in client factory- fixes #219

jest/README.md

@@ -452,10 +452,12 @@ SSLContext sslContext = new SSLContextBuilder().loadTrustMaterial(null, new Trus
 HostnameVerifier hostnameVerifier = NoopHostnameVerifier.INSTANCE;
 
 SSLConnectionSocketFactory sslSocketFactory = new SSLConnectionSocketFactory(sslContext, hostnameVerifier);
+SchemeIOSessionStrategy httpsIOSessionStrategy = SSLIOSessionStrategy(sslContext, hostnameVerifier);
 
 JestClientFactory factory = new JestClientFactory();
 factory.setHttpClientConfig(new HttpClientConfig.Builder("https://localhost:9200")
-                .sslSocketFactory(sslSocketFactory)
+                .sslSocketFactory(sslSocketFactory) // this only affects sync calls
+                .httpsIOSessionStrategy(httpsIOSessionStrategy) // this only affects async calls
                 .build()
 );
 ```

jest/src/main/java/io/searchbox/client/JestClientFactory.java

@@ -18,7 +18,15 @@
 import org.apache.http.impl.client.HttpClients;
 import org.apache.http.impl.conn.BasicHttpClientConnectionManager;
 import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
 import org.apache.http.impl.nio.client.HttpAsyncClients;
+import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
+import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
+import org.apache.http.impl.nio.reactor.IOReactorConfig;
+import org.apache.http.nio.conn.NHttpClientConnectionManager;
+import org.apache.http.nio.conn.SchemeIOSessionStrategy;
+import org.apache.http.nio.reactor.IOReactorException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -43,7 +51,9 @@ public JestClient getObject() {
         client.setRequestCompressionEnabled(httpClientConfig.isRequestCompressionEnabled());
         client.setServers(httpClientConfig.getServerList());
         final HttpClientConnectionManager connectionManager = getConnectionManager();
+        final NHttpClientConnectionManager asyncConnectionManager = getAsyncConnectionManager();
         client.setHttpClient(createHttpClient(connectionManager));
+        client.setAsyncClient(createAsyncHttpClient(asyncConnectionManager));
 
         // set custom gson instance
         Gson gson = httpClientConfig.getGson();
@@ -69,15 +79,14 @@ public JestClient getObject() {
         if (httpClientConfig.getMaxConnectionIdleTime() > 0) {
             log.info("Idle connection reaping enabled...");
 
-            IdleConnectionReaper reaper = new IdleConnectionReaper(httpClientConfig, new HttpReapableConnectionManager(connectionManager));
+            IdleConnectionReaper reaper = new IdleConnectionReaper(httpClientConfig, new HttpReapableConnectionManager(connectionManager, asyncConnectionManager));
             client.setIdleConnectionReaper(reaper);
             reaper.startAsync();
             reaper.awaitRunning();
         } else {
             log.info("Idle connection reaping disabled...");
         }
 
-        client.setAsyncClient(HttpAsyncClients.custom().setRoutePlanner(getRoutePlanner()).build());
         return client;
     }
 
@@ -96,10 +105,22 @@ private CloseableHttpClient createHttpClient(HttpClientConnectionManager connect
         ).build();
     }
 
+    private CloseableHttpAsyncClient createAsyncHttpClient(NHttpClientConnectionManager connectionManager) {
+        return configureHttpClient(
+                HttpAsyncClients.custom()
+                        .setConnectionManager(connectionManager)
+                        .setDefaultRequestConfig(getRequestConfig())
+                        .setProxyAuthenticationStrategy(httpClientConfig.getProxyAuthenticationStrategy())
+                        .setRoutePlanner(getRoutePlanner())
+                        .setDefaultCredentialsProvider(httpClientConfig.getCredentialsProvider())
+        ).build();
+    }
+
     /**
      * Extension point
-     * <p/>
+     * <p>
      * Example:
+     * </p>
      * <pre>
      * final JestClientFactory factory = new JestClientFactory() {
      *    {@literal @Override}
@@ -108,14 +129,18 @@ private CloseableHttpClient createHttpClient(HttpClientConnectionManager connect
      *    }
      * }
      * </pre>
-     *
-     * @param builder
-     * @return
      */
     protected HttpClientBuilder configureHttpClient(final HttpClientBuilder builder) {
         return builder;
     }
 
+    /**
+     * Extension point for async client
+     */
+    protected HttpAsyncClientBuilder configureHttpClient(final HttpAsyncClientBuilder builder) {
+        return builder;
+    }
+
     // Extension point
     protected HttpRoutePlanner getRoutePlanner() {
         return httpClientConfig.getHttpRoutePlanner();
@@ -129,6 +154,45 @@ protected RequestConfig getRequestConfig() {
                 .build();
     }
 
+    // Extension point
+    protected NHttpClientConnectionManager getAsyncConnectionManager() {
+        PoolingNHttpClientConnectionManager retval;
+
+        IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
+                .setConnectTimeout(httpClientConfig.getConnTimeout())
+                .setSoTimeout(httpClientConfig.getReadTimeout())
+                .build();
+
+        Registry<SchemeIOSessionStrategy> sessionStrategyRegistry = RegistryBuilder.<SchemeIOSessionStrategy>create()
+                .register("http", httpClientConfig.getHttpIOSessionStrategy())
+                .register("https", httpClientConfig.getHttpsIOSessionStrategy())
+                .build();
+
+        try {
+            retval = new PoolingNHttpClientConnectionManager(
+                    new DefaultConnectingIOReactor(ioReactorConfig),
+                    sessionStrategyRegistry
+            );
+        } catch (IOReactorException e) {
+            throw new IllegalStateException(e);
+        }
+
+        final Integer maxTotal = httpClientConfig.getMaxTotalConnection();
+        if (maxTotal != null) {
+            retval.setMaxTotal(maxTotal);
+        }
+        final Integer defaultMaxPerRoute = httpClientConfig.getDefaultMaxTotalConnectionPerRoute();
+        if (defaultMaxPerRoute != null) {
+            retval.setDefaultMaxPerRoute(defaultMaxPerRoute);
+        }
+        final Map<HttpRoute, Integer> maxPerRoute = httpClientConfig.getMaxTotalConnectionPerRoute();
+        for (Map.Entry<HttpRoute, Integer> entry : maxPerRoute.entrySet()) {
+            retval.setMaxPerRoute(entry.getKey(), entry.getValue());
+        }
+
+        return retval;
+    }
+
     // Extension point
     protected HttpClientConnectionManager getConnectionManager() {
         HttpClientConnectionManager retval;

jest/src/main/java/io/searchbox/client/config/HttpClientConfig.java

@@ -14,6 +14,9 @@
 import org.apache.http.impl.client.BasicCredentialsProvider;
 import org.apache.http.impl.conn.DefaultProxyRoutePlanner;
 import org.apache.http.impl.conn.SystemDefaultRoutePlanner;
+import org.apache.http.nio.conn.NoopIOSessionStrategy;
+import org.apache.http.nio.conn.SchemeIOSessionStrategy;
+import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
 
 import java.net.ProxySelector;
 import java.util.Collection;
@@ -34,6 +37,8 @@ public class HttpClientConfig extends ClientConfig {
     private final ConnectionSocketFactory plainSocketFactory;
     private final HttpRoutePlanner httpRoutePlanner;
     private final AuthenticationStrategy proxyAuthenticationStrategy;
+    private final SchemeIOSessionStrategy httpIOSessionStrategy;
+    private final SchemeIOSessionStrategy httpsIOSessionStrategy;
 
     public HttpClientConfig(Builder builder) {
         super(builder);
@@ -45,6 +50,8 @@ public HttpClientConfig(Builder builder) {
         this.plainSocketFactory = builder.plainSocketFactory;
         this.httpRoutePlanner = builder.httpRoutePlanner;
         this.proxyAuthenticationStrategy = builder.proxyAuthenticationStrategy;
+        this.httpIOSessionStrategy = builder.httpIOSessionStrategy;
+        this.httpsIOSessionStrategy = builder.httpsIOSessionStrategy;
     }
 
     public Map<HttpRoute, Integer> getMaxTotalConnectionPerRoute() {
@@ -79,6 +86,14 @@ public AuthenticationStrategy getProxyAuthenticationStrategy() {
         return proxyAuthenticationStrategy;
     }
 
+    public SchemeIOSessionStrategy getHttpIOSessionStrategy() {
+        return httpIOSessionStrategy;
+    }
+
+    public SchemeIOSessionStrategy getHttpsIOSessionStrategy() {
+        return httpsIOSessionStrategy;
+    }
+
     public static class Builder extends ClientConfig.AbstractBuilder<HttpClientConfig, Builder> {
 
         private Integer maxTotalConnection;
@@ -89,6 +104,8 @@ public static class Builder extends ClientConfig.AbstractBuilder<HttpClientConfi
         private ConnectionSocketFactory plainSocketFactory = PlainConnectionSocketFactory.getSocketFactory();
         private HttpRoutePlanner httpRoutePlanner = new SystemDefaultRoutePlanner(ProxySelector.getDefault());
         private AuthenticationStrategy proxyAuthenticationStrategy;
+        private SchemeIOSessionStrategy httpIOSessionStrategy = NoopIOSessionStrategy.INSTANCE;
+        private SchemeIOSessionStrategy httpsIOSessionStrategy = SSLIOSessionStrategy.getSystemDefaultStrategy();
 
         public Builder(HttpClientConfig httpClientConfig) {
             super(httpClientConfig);
@@ -144,21 +161,76 @@ public Builder defaultCredentials(String username, String password) {
         }
 
         /**
-         * @param socketFactory The socket factory instance that will be registered for <code>https</code> scheme.
+         * Sets the socket factory that will be used by <b>sync</b> client for HTTP scheme.
+         * <p>
+         * <code>SSLConnectionSocketFactory.getSocketFactory()</code> is used by default.
+         * </p><p>
+         * A bad example of trust-all socket factory creation can be done as below:
+         * </p>
+         * <pre>
+         * // trust ALL certificates
+         * SSLContext sslContext = new SSLContextBuilder().loadTrustMaterial(null, new TrustStrategy() {
+         *     public boolean isTrusted(X509Certificate[] arg0, String arg1) throws CertificateException {
+         *         return true;
+         *     }
+         * }).build();
+         *
+         * // skip hostname checks
+         * HostnameVerifier hostnameVerifier = NoopHostnameVerifier.INSTANCE;
+         *
+         * SSLConnectionSocketFactory sslSocketFactory = new SSLConnectionSocketFactory(sslContext, hostnameVerifier);
+         * </pre>
+         *
+         * @param socketFactory socket factory instance that will be registered for <code>https</code> scheme.
+         * @see SSLConnectionSocketFactory
          */
         public Builder sslSocketFactory(LayeredConnectionSocketFactory socketFactory) {
             this.sslSocketFactory = socketFactory;
             return this;
         }
 
         /**
-         * @param socketFactory The socket factory instance that will be registered for <code>http</code> scheme.
+         * Sets the socket factory that will be used by <b>sync</b> client for HTTPS scheme.
+         * <p>
+         * <code>PlainConnectionSocketFactory.getSocketFactory()</code> is used by default.
+         * </p>
+         *
+         * @param socketFactory socket factory instance that will be registered for <code>http</code> scheme.
+         * @see PlainConnectionSocketFactory
          */
         public Builder plainSocketFactory(ConnectionSocketFactory socketFactory) {
             this.plainSocketFactory = socketFactory;
             return this;
         }
 
+        /**
+         * Sets the socket factory that will be used by <b>async</b> client for HTTP scheme.
+         * <p>
+         * <code>NoopIOSessionStrategy.INSTANCE</code> is used by default.
+         * </p>
+         *
+         * @param httpIOSessionStrategy SchemeIOSessionStrategy instance that will be registered for <code>http</code> scheme.
+         * @see NoopIOSessionStrategy
+         */
+        public Builder httpIOSessionStrategy(SchemeIOSessionStrategy httpIOSessionStrategy) {
+            this.httpIOSessionStrategy = httpIOSessionStrategy;
+            return this;
+        }
+
+        /**
+         * Sets the socket factory that will be used by <b>async</b> client for HTTPS scheme.
+         * <p>
+         * <code>SSLIOSessionStrategy.getSystemDefaultStrategy()</code> is used by default.
+         * </p>
+         *
+         * @param httpsIOSessionStrategy SchemeIOSessionStrategy instance that will be registered for <code>https</code> scheme.
+         * @see SSLIOSessionStrategy
+         */
+        public Builder httpsIOSessionStrategy(SchemeIOSessionStrategy httpsIOSessionStrategy) {
+            this.httpsIOSessionStrategy = httpsIOSessionStrategy;
+            return this;
+        }
+
         public Builder proxy(HttpHost proxy) {
             return proxy(proxy, null);
         }

jest/src/main/java/io/searchbox/client/config/idle/HttpReapableConnectionManager.java

@@ -1,18 +1,24 @@
 package io.searchbox.client.config.idle;
 
 import org.apache.http.conn.HttpClientConnectionManager;
+import org.apache.http.nio.conn.NHttpClientConnectionManager;
 
 import java.util.concurrent.TimeUnit;
 
 public class HttpReapableConnectionManager implements ReapableConnectionManager {
     private final HttpClientConnectionManager connectionManager;
+    private final NHttpClientConnectionManager nConnectionManager;
+
+    public HttpReapableConnectionManager(HttpClientConnectionManager connectionManager, NHttpClientConnectionManager nConnectionManager) {
+        if(connectionManager == null || nConnectionManager == null) throw new IllegalArgumentException();
 
-    public HttpReapableConnectionManager(HttpClientConnectionManager connectionManager) {
         this.connectionManager = connectionManager;
+        this.nConnectionManager = nConnectionManager;
     }
 
     @Override
     public void closeIdleConnections(long idleTimeout, TimeUnit unit) {
         connectionManager.closeIdleConnections(idleTimeout, unit);
+        nConnectionManager.closeIdleConnections(idleTimeout, unit);
     }
 }

jest/src/test/java/io/searchbox/client/JestClientFactoryTest.java

@@ -8,6 +8,8 @@
 import org.apache.http.conn.routing.HttpRoute;
 import org.apache.http.impl.conn.BasicHttpClientConnectionManager;
 import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
+import org.apache.http.nio.conn.NHttpClientConnectionManager;
 import org.junit.Test;
 
 import static org.junit.Assert.*;
@@ -70,16 +72,21 @@ public void multiThreadedClientCreation() {
         JestHttpClient jestClient = (JestHttpClient) factory.getObject();
 
         assertTrue(jestClient != null);
-        assertNotNull(jestClient.getAsyncClient());
+        assertEquals(jestClient.getServerPoolSize(), 1);
+        assertEquals("server list should contain localhost:9200", "http://localhost:9200", jestClient.getNextServer());
+
         final HttpClientConnectionManager connectionManager = factory.getConnectionManager();
         assertTrue(connectionManager instanceof PoolingHttpClientConnectionManager);
         assertEquals(10, ((PoolingHttpClientConnectionManager) connectionManager).getDefaultMaxPerRoute());
         assertEquals(20, ((PoolingHttpClientConnectionManager) connectionManager).getMaxTotal());
         assertEquals(5, ((PoolingHttpClientConnectionManager) connectionManager).getMaxPerRoute(routeOne));
         assertEquals(6, ((PoolingHttpClientConnectionManager) connectionManager).getMaxPerRoute(routeTwo));
 
-        assertEquals(jestClient.getServerPoolSize(), 1);
-        assertEquals("server list should contain localhost:9200",
-                "http://localhost:9200", jestClient.getNextServer());
+        final NHttpClientConnectionManager nConnectionManager = factory.getAsyncConnectionManager();
+        assertTrue(nConnectionManager instanceof PoolingNHttpClientConnectionManager);
+        assertEquals(10, ((PoolingNHttpClientConnectionManager) nConnectionManager).getDefaultMaxPerRoute());
+        assertEquals(20, ((PoolingNHttpClientConnectionManager) nConnectionManager).getMaxTotal());
+        assertEquals(5, ((PoolingNHttpClientConnectionManager) nConnectionManager).getMaxPerRoute(routeOne));
+        assertEquals(6, ((PoolingNHttpClientConnectionManager) nConnectionManager).getMaxPerRoute(routeTwo));
     }
 }