Introduce a way to flush a ChannelPool partition, close AsyncHttpClient#679

Author: Stephane Landelle

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

@@ -25,6 +25,7 @@
 import org.asynchttpclient.ListenableFuture;
 import org.asynchttpclient.Request;
 import org.asynchttpclient.providers.netty.channel.ChannelManager;
+import org.asynchttpclient.providers.netty.channel.pool.ChannelPoolPartitionSelector;
 import org.asynchttpclient.providers.netty.request.NettyRequestSender;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -79,4 +80,12 @@ public void close() {
     public <T> ListenableFuture<T> execute(Request request, final AsyncHandler<T> asyncHandler) throws IOException {
         return requestSender.sendRequest(request, asyncHandler, null, false);
     }
+
+    public void flushChannelPoolPartition(String partitionId) {
+        channelManager.flushPartition(partitionId);
+    }
+
+    public void flushChannelPoolPartitions(ChannelPoolPartitionSelector selector) {
+        channelManager.flushPartitions(selector);
+    }
 }

providers/netty/src/main/java/org/asynchttpclient/providers/netty/channel/ChannelManager.java

@@ -50,6 +50,7 @@
 import org.asynchttpclient.providers.netty.Callback;
 import org.asynchttpclient.providers.netty.NettyAsyncHttpProviderConfig;
 import org.asynchttpclient.providers.netty.channel.pool.ChannelPool;
+import org.asynchttpclient.providers.netty.channel.pool.ChannelPoolPartitionSelector;
 import org.asynchttpclient.providers.netty.channel.pool.DefaultChannelPool;
 import org.asynchttpclient.providers.netty.channel.pool.NoopChannelPool;
 import org.asynchttpclient.providers.netty.future.NettyResponseFuture;
@@ -426,4 +427,12 @@ public void call() throws Exception {
     public void drainChannel(final Channel channel, final NettyResponseFuture<?> future) {
         Channels.setAttribute(channel, newDrainCallback(future, channel, future.isKeepAlive(), getPartitionId(future)));
     }
+
+    public void flushPartition(String partitionId) {
+        channelPool.flushPartition(partitionId);
+    } 
+
+    public void flushPartitions(ChannelPoolPartitionSelector selector) {
+        channelPool.flushPartitions(selector);
+    }
 }

providers/netty/src/main/java/org/asynchttpclient/providers/netty/channel/pool/ChannelPool.java

@@ -54,4 +54,18 @@ public interface ChannelPool {
      * Destroy all channels that has been cached by this instance.
      */
     void destroy();
+
+    /**
+     * Flush a partition
+     * 
+     * @param partitionId
+     */
+    void flushPartition(String partitionId);
+
+    /**
+     * Flush partitions based on a selector
+     * 
+     * @param selector
+     */
+    void flushPartitions(ChannelPoolPartitionSelector selector);
 }

providers/netty/src/main/java/org/asynchttpclient/providers/netty/channel/pool/ChannelPoolPartitionSelector.java

@@ -0,0 +1,19 @@
+/*
+ * Copyright (c) 2014 AsyncHttpClient Project. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at
+ *     http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package org.asynchttpclient.providers.netty.channel.pool;
+
+public interface ChannelPoolPartitionSelector {
+
+   boolean select(String partitionId); 
+}

providers/netty/src/main/java/org/asynchttpclient/providers/netty/channel/pool/DefaultChannelPool.java

@@ -23,6 +23,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.TimeUnit;
@@ -320,4 +321,27 @@ private void close(Channel channel) {
         channel2Creation.remove(channel);
         Channels.silentlyCloseChannel(channel);
     }
+
+    private void flushPartition(String partitionId, ConcurrentLinkedQueue<IdleChannel> partition) {
+        if (partition != null) {
+            partitions.remove(partitionId);
+            for (IdleChannel idleChannel : partition)
+                close(idleChannel.channel);
+        }
+    }
+    
+    @Override
+    public void flushPartition(String partitionId) {
+        flushPartition(partitionId, partitions.get(partitionId));
+    }
+    
+    @Override
+    public void flushPartitions(ChannelPoolPartitionSelector selector) {
+
+        for (Map.Entry<String, ConcurrentLinkedQueue<IdleChannel>> partitionsEntry : partitions.entrySet()) {
+            String partitionId = partitionsEntry.getKey();
+            if (selector.select(partitionId))
+                flushPartition(partitionId, partitionsEntry.getValue());
+        }
+    }
 }

providers/netty/src/main/java/org/asynchttpclient/providers/netty/channel/pool/NoopChannelPool.java

@@ -17,22 +17,35 @@
 
 public class NoopChannelPool implements ChannelPool {
 
+    @Override
     public boolean offer(Channel channel, String poolKey) {
         return false;
     }
 
+    @Override
     public Channel poll(String poolKey) {
         return null;
     }
 
+    @Override
     public boolean removeAll(Channel channel) {
         return false;
     }
 
+    @Override
     public boolean isOpen() {
         return true;
     }
 
+    @Override
     public void destroy() {
     }
+
+    @Override
+    public void flushPartition(String partitionId) {
+    }
+
+    @Override
+    public void flushPartitions(ChannelPoolPartitionSelector selector) {
+    }
 }

providers/netty/src/test/java/org/asynchttpclient/providers/netty/NettyConnectionPoolTest.java

@@ -14,19 +14,19 @@
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
+import io.netty.channel.Channel;
+
+import java.net.ConnectException;
+import java.util.concurrent.TimeUnit;
 
 import org.asynchttpclient.AsyncHttpClient;
 import org.asynchttpclient.AsyncHttpClientConfig;
 import org.asynchttpclient.Response;
 import org.asynchttpclient.async.ConnectionPoolTest;
 import org.asynchttpclient.providers.netty.channel.pool.ChannelPool;
+import org.asynchttpclient.providers.netty.channel.pool.NoopChannelPool;
 import org.testng.annotations.Test;
 
-import io.netty.channel.Channel;
-
-import java.net.ConnectException;
-import java.util.concurrent.TimeUnit;
-
 public class NettyConnectionPoolTest extends ConnectionPoolTest {
 
     @Override
@@ -36,27 +36,17 @@ public AsyncHttpClient getAsyncHttpClient(AsyncHttpClientConfig config) {
 
     @Test(groups = { "standalone", "default_provider" })
     public void testInvalidConnectionsPool() {
-        ChannelPool cp = new ChannelPool() {
+        ChannelPool cp = new NoopChannelPool() {
 
-            public boolean offer(Channel channel, String poolKey) {
-                return false;
-            }
-
-            public Channel poll(String poolKey) {
-                return null;
-            }
-
-            public boolean removeAll(Channel channel) {
+            @Override
+            public boolean offer(Channel connection, String poolKey) {
                 return false;
             }
 
+            @Override
             public boolean isOpen() {
                 return false;
             }
-
-            public void destroy() {
-
-            }
         };
 
         NettyAsyncHttpProviderConfig providerConfig = new NettyAsyncHttpProviderConfig();
@@ -80,27 +70,12 @@ public void destroy() {
 
     @Test(groups = { "standalone", "default_provider" })
     public void testValidConnectionsPool() {
-        ChannelPool cp = new ChannelPool() {
-
-            public boolean offer(Channel channel, String poolKey) {
-                return true;
-            }
-
-            public Channel poll(String poolKey) {
-                return null;
-            }
+        ChannelPool cp = new NoopChannelPool() {
 
-            public boolean removeAll(Channel channel) {
-                return false;
-            }
-
-            public boolean isOpen() {
+            @Override
+            public boolean offer(Channel connection, String poolKey) {
                 return true;
             }
-
-            public void destroy() {
-
-            }
         };
 
         NettyAsyncHttpProviderConfig providerConfig = new NettyAsyncHttpProviderConfig();