Merge pull request AsyncHttpClient#134 from kstokoz/master

An option to limit pooled connections life time

Author: jfarcand

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

@@ -108,6 +108,7 @@ public class AsyncHttpClientConfig {
     protected HostnameVerifier hostnameVerifier;
     protected int ioThreadMultiplier;
     protected boolean strict302Handling;
+    protected int maxConnectionLifeTimeInMs;
 
     protected AsyncHttpClientConfig() {
     }
@@ -119,6 +120,7 @@ private AsyncHttpClientConfig(int maxTotalConnections,
                                   int idleConnectionInPoolTimeoutInMs,
                                   int idleConnectionTimeoutInMs,
                                   int requestTimeoutInMs,
+                                  int connectionMaxLifeTimeInMs,
                                   boolean redirectEnabled,
                                   int maxDefaultRedirects,
                                   boolean compressionEnabled,
@@ -150,6 +152,7 @@ private AsyncHttpClientConfig(int maxTotalConnections,
         this.idleConnectionInPoolTimeoutInMs = idleConnectionInPoolTimeoutInMs;
         this.idleConnectionTimeoutInMs = idleConnectionTimeoutInMs;
         this.requestTimeoutInMs = requestTimeoutInMs;
+        this.maxConnectionLifeTimeInMs = connectionMaxLifeTimeInMs;
         this.redirectEnabled = redirectEnabled;
         this.maxDefaultRedirects = maxDefaultRedirects;
         this.compressionEnabled = compressionEnabled;
@@ -500,6 +503,15 @@ public boolean isStrict302Handling() {
         return strict302Handling;
     }
 
+    /**
+     * Return the maximum time in millisecond an {@link com.ning.http.client.AsyncHttpClient} will keep connection in the pool, or -1 to keep connection while possible.
+     *
+     * @return the maximum time in millisecond an {@link com.ning.http.client.AsyncHttpClient} will keep connection in the pool, or -1 to keep connection while possible.
+     */
+    public int getMaxConnectionLifeTimeInMs() {
+        return maxConnectionLifeTimeInMs;
+    }
+
     /**
      * Builder for an {@link AsyncHttpClient}
      */
@@ -511,6 +523,7 @@ public static class Builder {
         private int defaultIdleConnectionInPoolTimeoutInMs = Integer.getInteger(ASYNC_CLIENT + "defaultIdleConnectionInPoolTimeoutInMS", 60 * 1000);
         private int defaultIdleConnectionTimeoutInMs = Integer.getInteger(ASYNC_CLIENT + "defaultIdleConnectionTimeoutInMS", 60 * 1000);
         private int defaultRequestTimeoutInMs = Integer.getInteger(ASYNC_CLIENT + "defaultRequestTimeoutInMS", 60 * 1000);
+        private int defaultMaxConnectionLifeTimeInMs = Integer.getInteger(ASYNC_CLIENT + "defaultMaxConnectionLifeTimeInMs", -1);
         private boolean redirectEnabled = Boolean.getBoolean(ASYNC_CLIENT + "defaultRedirectsEnabled");
         private int maxDefaultRedirects = Integer.getInteger(ASYNC_CLIENT + "defaultMaxRedirects", 5);
         private boolean compressionEnabled = Boolean.getBoolean(ASYNC_CLIENT + "compressionEnabled");
@@ -980,6 +993,17 @@ public Builder setStrict302Handling(final boolean strict302Handling) {
             return this;
         }
 
+        /**
+         * Set the maximum time in millisecond connection can be added to the pool for further reuse
+         *
+         * @param maxConnectionLifeTimeInMs the maximum time in millisecond connection can be added to the pool for further reuse
+         * @return a {@link Builder}
+         */
+        public Builder setMaxConnectionLifeTimeInMs(int maxConnectionLifeTimeInMs) {
+           this.defaultMaxConnectionLifeTimeInMs = maxConnectionLifeTimeInMs;
+           return this;
+        }
+
         /**
          * Create a config builder with values taken from the given prototype configuration.
          *
@@ -993,6 +1017,7 @@ public Builder(AsyncHttpClientConfig prototype) {
             defaultIdleConnectionInPoolTimeoutInMs = prototype.getIdleConnectionInPoolTimeoutInMs();
             defaultIdleConnectionTimeoutInMs = prototype.getIdleConnectionTimeoutInMs();
             defaultMaxConnectionPerHost = prototype.getMaxConnectionPerHost();
+            defaultMaxConnectionLifeTimeInMs = prototype.getMaxConnectionLifeTimeInMs();
             maxDefaultRedirects = prototype.getMaxRedirects();
             defaultMaxTotalConnections = prototype.getMaxTotalConnections();
             proxyServer = prototype.getProxyServer();
@@ -1046,6 +1071,7 @@ public AsyncHttpClientConfig build() {
                     defaultIdleConnectionInPoolTimeoutInMs,
                     defaultIdleConnectionTimeoutInMs,
                     defaultRequestTimeoutInMs,
+                    defaultMaxConnectionLifeTimeInMs,
                     redirectEnabled,
                     maxDefaultRedirects,
                     compressionEnabled,

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

@@ -34,18 +34,21 @@ public class NettyConnectionsPool implements ConnectionsPool<String, Channel> {
     private final static Logger log = LoggerFactory.getLogger(NettyConnectionsPool.class);
     private final ConcurrentHashMap<String, ConcurrentLinkedQueue<IdleChannel>> connectionsPool = new ConcurrentHashMap<String, ConcurrentLinkedQueue<IdleChannel>>();
     private final ConcurrentHashMap<Channel, IdleChannel> channel2IdleChannel = new ConcurrentHashMap<Channel, IdleChannel>();
+    private final ConcurrentHashMap<Channel, Long> channel2CreationDate = new ConcurrentHashMap<Channel, Long>();
     private final AtomicBoolean isClosed = new AtomicBoolean(false);
     private final Timer idleConnectionDetector = new Timer(true);
     private final boolean sslConnectionPoolEnabled;
     private final int maxTotalConnections;
     private final int maxConnectionPerHost;
+    private final int maxConnectionLifeTimeInMs;
     private final long maxIdleTime;
 
     public NettyConnectionsPool(NettyAsyncHttpProvider provider) {
         this.maxTotalConnections = provider.getConfig().getMaxTotalConnections();
         this.maxConnectionPerHost = provider.getConfig().getMaxConnectionPerHost();
         this.sslConnectionPoolEnabled = provider.getConfig().isSslConnectionPoolEnabled();
         this.maxIdleTime = provider.getConfig().getIdleConnectionInPoolTimeoutInMs();
+        this.maxConnectionLifeTimeInMs = provider.getConfig().getMaxConnectionLifeTimeInMs();
         this.idleConnectionDetector.schedule(new IdleChannelDetector(), maxIdleTime, maxIdleTime);
     }
 
@@ -145,6 +148,15 @@ public boolean offer(String uri, Channel channel) {
             return false;
         }
 
+        Long createTime = channel2CreationDate.get(channel);
+        if (createTime == null){
+           channel2CreationDate.putIfAbsent(channel, System.currentTimeMillis());
+        }
+        else if (maxConnectionLifeTimeInMs != -1 && (createTime + maxConnectionLifeTimeInMs) < System.currentTimeMillis() ) {
+           log.debug("Channel {} expired", channel);
+           return false;
+        }
+
         log.debug("Adding uri: {} for channel {}", uri, channel);
         channel.getPipeline().getContext(NettyAsyncHttpProvider.class).setAttachment(new NettyAsyncHttpProvider.DiscardEvent());
 
@@ -222,6 +234,7 @@ private boolean remove(IdleChannel pooledChannel) {
      * {@inheritDoc}
      */
     public boolean removeAll(Channel channel) {
+        channel2CreationDate.remove(channel);
         return !isClosed.get() && remove(channel2IdleChannel.get(channel));
     }
 
@@ -251,11 +264,13 @@ public void destroy() {
         }
         connectionsPool.clear();
         channel2IdleChannel.clear();
+        channel2CreationDate.clear();
     }
 
     private void close(Channel channel) {
         try {
             channel.getPipeline().getContext(NettyAsyncHttpProvider.class).setAttachment(new NettyAsyncHttpProvider.DiscardEvent());
+            channel2CreationDate.remove(channel);
             channel.close();
         } catch (Throwable t) {
             // noop