Do not create a thread pool in AsyncHttpClientConfig.

Developers may still specify a service that they may wish to share with the AHC instance.

By having this not be configured by default, it will allow providers to provide more optimized configurations out of the box.

Author: rlubke

api/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java

@@ -106,7 +106,6 @@ public class AsyncHttpClientConfig {
     protected boolean allowSslConnectionPool;
     protected boolean useRawUrl;
     protected boolean removeQueryParamOnRedirect;
-    protected boolean managedApplicationThreadPool;
     protected HostnameVerifier hostnameVerifier;
     protected int ioThreadMultiplier;
     protected boolean strict302Handling;
@@ -157,8 +156,7 @@ private AsyncHttpClientConfig(int maxTotalConnections,
                                   int spdyInitialWindowSize,
                                   int spdyMaxConcurrentStreams,
                                   boolean rfc6265CookieEncoding,
-                                  boolean asyncConnectMode,
-                                  boolean managedApplicationThreadPool) {
+                                  boolean asyncConnectMode) {
 
         this.maxTotalConnections = maxTotalConnections;
         this.maxConnectionPerHost = maxConnectionPerHost;
@@ -190,7 +188,6 @@ private AsyncHttpClientConfig(int maxTotalConnections,
         this.ioThreadMultiplier = ioThreadMultiplier;
         this.strict302Handling = strict302Handling;
         this.useRelativeURIsWithSSLProxies = useRelativeURIsWithSSLProxies;
-        this.managedApplicationThreadPool = managedApplicationThreadPool;
         this.applicationThreadPool = applicationThreadPool;
         this.proxyServerSelector = proxyServerSelector;
         this.useRawUrl = useRawUrl;
@@ -334,25 +331,13 @@ public boolean isCompressionEnabled() {
      * asynchronous response.
      *
      * @return the {@link java.util.concurrent.ExecutorService} an {@link AsyncHttpClient} use for handling
-     *         asynchronous response.
+     *         asynchronous response.  If no {@link ExecutorService} has been explicitly provided,
+     *         this method will return <code>null</code>
      */
     public ExecutorService executorService() {
         return applicationThreadPool;
     }
 
-    /**
-     * @return <code>true</code> if this <code>AsyncHttpClientConfig</code> instance created the
-     *  {@link ExecutorService} returned by {@link #executorService()}, otherwise returns <code>false</code>.
-     *  The return from this method is typically used by the various provider implementations to determine
-     *  if it should shutdown the {@link ExecutorService} when the {@link AsyncHttpClient} is closed.  Developers
-     *  should take care and not share managed {@link ExecutorService} instances between client instances.
-     *
-     * @since 2.2.0
-     */
-    public boolean isManagedExecutorService() {
-        return managedApplicationThreadPool;
-    }
-
     /**
      * An instance of {@link ProxyServer} used by an {@link AsyncHttpClient}
      *
@@ -631,7 +616,6 @@ public static class Builder {
         private boolean useRelativeURIsWithSSLProxies = Boolean.getBoolean(ASYNC_CLIENT + "useRelativeURIsWithSSLProxies");
         private ScheduledExecutorService reaper;
         private ExecutorService applicationThreadPool;
-        private boolean managedApplicationThreadPool;
         private ProxyServerSelector proxyServerSelector = null;
         private SSLContext sslContext;
         private SSLEngineFactory sslEngineFactory;
@@ -1272,23 +1256,22 @@ public Thread newThread(Runnable r) {
                 });
             }
 
-            if (applicationThreadPool == null) {
-                managedApplicationThreadPool = true;
-                applicationThreadPool =
-                        Executors.newCachedThreadPool(new ThreadFactory() {
-                            final AtomicInteger counter = new AtomicInteger();
-                            public Thread newThread(Runnable r) {
-                                Thread t = new Thread(r,
-                                                      "AsyncHttpClient-Callback-" + counter.incrementAndGet());
-                                t.setDaemon(true);
-                                return t;
-                            }
-                        });
-            }
-
-            if (applicationThreadPool.isShutdown()) {
-                throw new IllegalStateException("ExecutorServices closed");
-            }
+//            if (applicationThreadPool == null) {
+//                applicationThreadPool =
+//                        Executors.newCachedThreadPool(new ThreadFactory() {
+//                            final AtomicInteger counter = new AtomicInteger();
+//                            public Thread newThread(Runnable r) {
+//                                Thread t = new Thread(r,
+//                                                      "AsyncHttpClient-Callback-" + counter.incrementAndGet());
+//                                t.setDaemon(true);
+//                                return t;
+//                            }
+//                        });
+//            }
+//
+//            if (applicationThreadPool.isShutdown()) {
+//                throw new IllegalStateException("ExecutorServices closed");
+//            }
 
             if (proxyServerSelector == null && useProxySelector) {
                 proxyServerSelector = ProxyUtils.getJdkDefaultProxyServerSelector();
@@ -1339,8 +1322,7 @@ public Thread newThread(Runnable r) {
                     spdyInitialWindowSize,
                     spdyMaxConcurrentStreams,
                     rfc6265CookieEncoding,
-                    asyncConnectMode,
-                    managedApplicationThreadPool);
+                    asyncConnectMode);
         }
     }
 }

api/src/main/java/org/asynchttpclient/providers/jdk/JDKAsyncHttpProvider.java

@@ -72,6 +72,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -98,9 +99,18 @@ public class JDKAsyncHttpProvider implements AsyncHttpProvider {
 
     private boolean bufferResponseInMemory = false;
 
+    private ExecutorService service;
+
+    private boolean managedExecutorService;
+
     public JDKAsyncHttpProvider(AsyncHttpClientConfig config) {
 
         this.config = config;
+        service = config.executorService();
+        managedExecutorService = (service == null);
+        if (service == null) {
+            service = AsyncHttpProviderUtils.createDefaultExecutorService();
+        }
         AsyncHttpProviderConfig<?, ?> providerConfig = config.getAsyncHttpProviderConfig();
         if (providerConfig instanceof JDKAsyncHttpProviderConfig) {
             configure(JDKAsyncHttpProviderConfig.class.cast(providerConfig));
@@ -152,7 +162,7 @@ public <T> ListenableFuture<T> execute(Request request, AsyncHandler<T> handler,
         JDKFuture<T> f = (delegate == null) ? new JDKFuture<T>(handler, requestTimeout, urlConnection) : delegate;
         f.touch();
 
-        f.setInnerFuture(config.executorService().submit(new AsyncHttpUrlConnection<T>(urlConnection, request, handler, f)));
+        f.setInnerFuture(service.submit(new AsyncHttpUrlConnection<T>(urlConnection, request, handler, f)));
         maxConnections.incrementAndGet();
 
         return f;
@@ -193,6 +203,9 @@ private HttpURLConnection createUrlConnection(Request request) throws IOExceptio
 
     public void close() {
         isClose.set(true);
+        if (managedExecutorService) {
+            service.shutdownNow();
+        }
     }
 
     public Response prepareResponse(HttpResponseStatus status, HttpResponseHeaders headers, List<HttpResponseBodyPart> bodyParts) {

api/src/main/java/org/asynchttpclient/util/AsyncHttpProviderUtils.java

@@ -28,6 +28,10 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Vector;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.asynchttpclient.AsyncHttpClientConfig;
 import org.asynchttpclient.AsyncHttpProvider;
@@ -582,4 +586,17 @@ public static String keepAliveHeaderValue(AsyncHttpClientConfig config) {
     public static int requestTimeout(AsyncHttpClientConfig config, Request request) {
         return request.getRequestTimeoutInMs() != 0 ? request.getRequestTimeoutInMs() : config.getRequestTimeoutInMs();
     }
+
+    public static ExecutorService createDefaultExecutorService() {
+        return Executors.newCachedThreadPool(new ThreadFactory() {
+            final AtomicInteger counter = new AtomicInteger();
+
+            public Thread newThread(Runnable r) {
+                Thread t = new Thread(r,
+                                      "AsyncHttpClient-Callback-" + counter.incrementAndGet());
+                t.setDaemon(true);
+                return t;
+            }
+        });
+    }
 }

providers/grizzly/src/main/java/org/asynchttpclient/providers/grizzly/GrizzlyAsyncHttpProvider.java

@@ -183,13 +183,11 @@ public void close() {
         try {
             connectionManager.destroy();
             clientTransport.stop();
-            if (clientConfig.isManagedExecutorService()) {
-                final ExecutorService service = clientConfig.executorService();
-                // service may be null due to a custom configuration that
-                // leverages Grizzly's SameThreadIOStrategy.
-                if (service != null) {
-                    service.shutdown();
-                }
+            final ExecutorService service = clientConfig.executorService();
+            // service may be null due to a custom configuration that
+            // leverages Grizzly's SameThreadIOStrategy.
+            if (service != null) {
+                service.shutdown();
             }
             if (timeoutExecutor != null) {
                 timeoutExecutor.stop();

providers/netty/src/main/java/org/asynchttpclient/providers/netty/NettyAsyncHttpProvider.java

@@ -132,9 +132,11 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.asynchttpclient.util.AsyncHttpProviderUtils.DEFAULT_CHARSET;
 import static org.asynchttpclient.util.DateUtil.millisTime;
@@ -183,6 +185,8 @@ public boolean remove(Object o) {
     private static SpnegoEngine spnegoEngine = null;
     private final Protocol httpProtocol = new HttpProtocol();
     private final Protocol webSocketProtocol = new WebSocketProtocol();
+    private final boolean managedExecutorService;
+    private ExecutorService service;
 
 	private static boolean isNTLM(List<String> auth) {
 		return isNonEmpty(auth) && auth.get(0).startsWith("NTLM");
@@ -195,9 +199,14 @@ public NettyAsyncHttpProvider(AsyncHttpClientConfig config) {
         } else {
             asyncHttpProviderConfig = new NettyAsyncHttpProviderConfig();
         }
+        service = config.executorService();
+        managedExecutorService = (service == null);
+        if (service == null) {
+            service = AsyncHttpProviderUtils.createDefaultExecutorService();
+        }
 
         if (asyncHttpProviderConfig.isUseBlockingIO()) {
-            socketChannelFactory = new OioClientSocketChannelFactory(config.executorService());
+            socketChannelFactory = new OioClientSocketChannelFactory(service);
             this.allowReleaseSocketChannelFactory = true;
         } else {
             // check if external NioClientSocketChannelFactory is defined
@@ -214,7 +223,7 @@ public NettyAsyncHttpProvider(AsyncHttpClientConfig config) {
                 }
                 int numWorkers = config.getIoThreadMultiplier() * Runtime.getRuntime().availableProcessors();
                 log.debug("Number of application's worker threads is {}", numWorkers);
-                socketChannelFactory = new NioClientSocketChannelFactory(e, config.executorService(), numWorkers);
+                socketChannelFactory = new NioClientSocketChannelFactory(e, service, numWorkers);
                 this.allowReleaseSocketChannelFactory = true;
             }
         }
@@ -830,8 +839,8 @@ public void close() {
                 }
             }
 
-            if (config.isManagedExecutorService()) {
-                config.executorService().shutdown();
+            if (managedExecutorService) {
+                service.shutdown();
             }
             config.reaper().shutdown();
             if (this.allowReleaseSocketChannelFactory) {