Move reaper to Apache provider specific config, close AsyncHttpClient#331

Author: Stephane Landelle

src/main/java/com/ning/http/client/AsyncHttpClientConfig.java

@@ -65,7 +65,6 @@ public class AsyncHttpClientConfig {
     protected boolean compressionEnabled;
     protected String userAgent;
     protected boolean allowPoolingConnection;
-    protected ScheduledExecutorService reaper;
     protected ExecutorService applicationThreadPool;
     protected ProxyServerSelector proxyServerSelector;
     protected SSLContext sslContext;
@@ -148,7 +147,6 @@ private AsyncHttpClientConfig(int maxTotalConnections,
         this.ioExceptionFilters = ioExceptionFilters;
         this.requestCompressionLevel = requestCompressionLevel;
         this.maxRequestRetry = maxRequestRetry;
-        this.reaper = reaper;
         this.allowSslConnectionPool = allowSslConnectionCaching;
         this.removeQueryParamOnRedirect = removeQueryParamOnRedirect;
         this.hostnameVerifier = hostnameVerifier;
@@ -166,15 +164,6 @@ private AsyncHttpClientConfig(int maxTotalConnections,
         this.useRawUrl = useRawUrl;
     }
 
-    /**
-     * A {@link ScheduledExecutorService} used to expire idle connections.
-     *
-     * @return {@link ScheduledExecutorService}
-     */
-    public ScheduledExecutorService reaper() {
-        return reaper;
-    }
-
     /**
      * Return the maximum number of connections an {@link com.ning.http.client.AsyncHttpClient} can handle.
      *
@@ -469,7 +458,7 @@ public boolean isValid() {
             // when using a ManagedExecutorService.
             // When this is the case, we assume it's running.
         }
-        return (atpRunning && !reaper.isShutdown());
+        return atpRunning;
     }
 
     /**
@@ -1091,7 +1080,6 @@ public Builder(AsyncHttpClientConfig prototype) {
             userAgent = prototype.getUserAgent();
             redirectEnabled = prototype.isRedirectEnabled();
             compressionEnabled = prototype.isCompressionEnabled();
-            reaper = prototype.reaper();
             applicationThreadPool = prototype.executorService();
 
             requestFilters.clear();

src/main/java/com/ning/http/client/AsyncHttpClientConfigBean.java

@@ -23,7 +23,6 @@
 import java.util.LinkedList;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 
 /**
@@ -79,13 +78,6 @@ public boolean verify(String s, SSLSession sslSession) {
     }
 
     void configureExecutors() {
-        reaper = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {
-            public Thread newThread(Runnable r) {
-                Thread t = new Thread(r, "AsyncHttpClient-Reaper");
-                t.setDaemon(true);
-                return t;
-            }
-        });
         applicationThreadPool = Executors.newCachedThreadPool(new ThreadFactory() {
             public Thread newThread(Runnable r) {
                 Thread t = new Thread(r, "AsyncHttpClient-Callback");
@@ -150,14 +142,6 @@ public AsyncHttpClientConfigBean setAllowPoolingConnection(boolean allowPoolingC
         return this;
     }
 
-    public AsyncHttpClientConfigBean setReaper(ScheduledExecutorService reaper) {
-        if (this.reaper != null) {
-            this.reaper.shutdownNow();
-        }
-        this.reaper = reaper;
-        return this;
-    }
-
     public AsyncHttpClientConfigBean setApplicationThreadPool(ExecutorService applicationThreadPool) {
         if (this.applicationThreadPool != null) {
             this.applicationThreadPool.shutdownNow();

src/main/java/com/ning/http/client/providers/apache/ApacheAsyncHttpProvider.java

@@ -45,6 +45,7 @@
 import com.ning.http.util.AsyncHttpProviderUtils;
 import com.ning.http.util.ProxyUtils;
 import com.ning.http.util.UTF8UrlEncoder;
+
 import org.apache.commons.httpclient.CircularRedirectException;
 import org.apache.commons.httpclient.Credentials;
 import org.apache.commons.httpclient.DefaultHttpMethodRetryHandler;
@@ -85,6 +86,7 @@
 import javax.net.ssl.SSLSocketFactory;
 import javax.net.ssl.TrustManager;
 import javax.net.ssl.X509TrustManager;
+
 import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.FileInputStream;
@@ -107,7 +109,10 @@
 import java.util.Map;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -129,6 +134,7 @@ public class ApacheAsyncHttpProvider implements AsyncHttpProvider {
     private final AtomicInteger maxConnections = new AtomicInteger();
     private final MultiThreadedHttpConnectionManager connectionManager;
     private final HttpClientParams params;
+    private final ScheduledExecutorService reaper;
 
     static {
         final SocketFactory factory = new TrustingSSLSocketFactory();
@@ -157,13 +163,26 @@ public ApacheAsyncHttpProvider(AsyncHttpClientConfig config) {
         params.setCookiePolicy(CookiePolicy.BROWSER_COMPATIBILITY);
         params.setParameter(HttpMethodParams.RETRY_HANDLER, new DefaultHttpMethodRetryHandler());
 
-        AsyncHttpProviderConfig<?, ?> providerConfig = config.getAsyncHttpProviderConfig();
+        reaper = getReaper(config.getAsyncHttpProviderConfig());
+    }
+
+    private ScheduledExecutorService getReaper(AsyncHttpProviderConfig<?, ?> providerConfig) {
+
+        ScheduledExecutorService reaper = null;
         if (providerConfig instanceof ApacheAsyncHttpProvider) {
-            configure(ApacheAsyncHttpProviderConfig.class.cast(providerConfig));
+            reaper = ApacheAsyncHttpProviderConfig.class.cast(providerConfig).getReaper();
         }
-    }
 
-    private void configure(ApacheAsyncHttpProviderConfig config) {
+        if (reaper == null)
+            reaper = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {
+                public Thread newThread(Runnable r) {
+                    Thread t = new Thread(r, "AsyncHttpClient-Reaper");
+                    t.setDaemon(true);
+                    return t;
+                }
+            });
+
+        return reaper;
     }
 
     public <T> ListenableFuture<T> execute(Request request, AsyncHandler<T> handler) throws IOException {
@@ -211,6 +230,7 @@ public <T> ListenableFuture<T> execute(Request request, AsyncHandler<T> handler)
     }
 
     public void close() {
+        reaper.shutdown();
         if (idleConnectionTimeoutThread != null) {
             idleConnectionTimeoutThread.shutdown();
             idleConnectionTimeoutThread = null;
@@ -455,7 +475,7 @@ public T call() {
                 int delay = requestTimeout(config, future.getRequest().getPerRequestConfig());
                 if (delay != -1) {
                     ReaperFuture reaperFuture = new ReaperFuture(future);
-                    Future scheduledFuture = config.reaper().scheduleAtFixedRate(reaperFuture, delay, 500, TimeUnit.MILLISECONDS);
+                    Future scheduledFuture = reaper.scheduleAtFixedRate(reaperFuture, delay, 500, TimeUnit.MILLISECONDS);
                     reaperFuture.setScheduledFuture(scheduledFuture);
                     future.setReaperFuture(reaperFuture);
                 }

src/main/java/com/ning/http/client/providers/apache/ApacheAsyncHttpProviderConfig.java

@@ -12,16 +12,18 @@
  */
 package com.ning.http.client.providers.apache;
 
-import com.ning.http.client.AsyncHttpProviderConfig;
-
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+
+import com.ning.http.client.AsyncHttpProviderConfig;
 
 public class ApacheAsyncHttpProviderConfig implements AsyncHttpProviderConfig<String, String> {
 
     private final ConcurrentHashMap<String, String> properties = new ConcurrentHashMap<String, String>();
 
+    private ScheduledExecutorService reaper;
 
     public AsyncHttpProviderConfig addProperty(String name, String value) {
         properties.put(name, value);
@@ -39,4 +41,12 @@ public String removeProperty(String name) {
     public Set<Map.Entry<String, String>> propertiesSet() {
         return properties.entrySet();
     }
+
+    public void setReaper(ScheduledExecutorService reaper) {
+        this.reaper = reaper;
+    }
+
+    public ScheduledExecutorService getReaper() {
+        return reaper;
+    }
 }

src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java

@@ -889,7 +889,6 @@ public void close() {
                 }
 
                 config.executorService().shutdown();
-                config.reaper().shutdown();
                 if (this.allowReleaseSocketChannelFactory) {
                     socketChannelFactory.releaseExternalResources();
                     plainBootstrap.releaseExternalResources();