Added an option for netty connection pool to limit the time when connection can be put back to pool.

In case of Amazon Cloud DNS records can change in seconds, and taking into account that AHC caches already resolved connection and never re-resolve DNS records if response if ok, introducing option to limit life for cached connection, to be able to gracefully route traffic to new host in case dns records changed.

Author: Kyrylo Stokoz

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