Expose flushPartition on AsyncHttpClient, close #1375

AsyncHttpClient · Mar 28, 2017 · fc7024c · fc7024c
1 parent eff9e93
commit fc7024c
Show file tree

Hide file tree

Showing 8 changed files with 39 additions and 23 deletions.
diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClient.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClient.java
@@ -18,6 +18,7 @@
 
 import java.io.Closeable;
 import java.util.concurrent.Future;
+import java.util.function.Predicate;
 
 /**
  * This class support asynchronous and synchronous HTTP request.
@@ -273,4 +274,11 @@ public interface AsyncHttpClient extends Closeable {
      * @return a {@link ClientStats}
      */
     ClientStats getClientStats();
+
+    /**
+     * Flush ChannelPool partitions based on a predicate
+     * 
+     * @param predicate the predicate
+     */
+    void flushChannelPoolPartitions(Predicate<Object> predicate);
 }
diff --git a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java
@@ -22,6 +22,7 @@
 import io.netty.util.Timer;
 
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Predicate;
 
 import org.asynchttpclient.channel.ChannelPool;
 import org.asynchttpclient.filter.FilterContext;
@@ -259,6 +260,11 @@ public EventLoopGroup getEventLoopGroup() {
     public ClientStats getClientStats() {
         return channelManager.getClientStats();
     }
+
+    @Override
+    public void flushChannelPoolPartitions(Predicate<Object> predicate) {
+        channelManager.flushChannelPoolPartitions(predicate);
+    }
 
     protected BoundRequestBuilder requestBuilder(String method, String url) {
         return new BoundRequestBuilder(this, method, config.isDisableUrlEncodingForBoundRequests()).setUrl(url).setSignatureCalculator(signatureCalculator);

diff --git a/client/src/main/java/org/asynchttpclient/channel/ChannelPool.java b/client/src/main/java/org/asynchttpclient/channel/ChannelPool.java
@@ -61,18 +61,11 @@ public interface ChannelPool {
     void destroy();
 
     /**
-     * Flush a partition
+     * Flush partitions based on a predicate
      * 
-     * @param partitionKey the partition
+     * @param predicate the predicate
      */
-    void flushPartition(Object partitionKey);
-
-    /**
-     * Flush partitions based on a selector
-     * 
-     * @param selector the selector
-     */
-    void flushPartitions(Predicate<Object> selector);
+    void flushPartitions(Predicate<Object> predicate);
 
     /**
      * @return The number of idle channels per host.

diff --git a/client/src/main/java/org/asynchttpclient/channel/NoopChannelPool.java b/client/src/main/java/org/asynchttpclient/channel/NoopChannelPool.java
@@ -48,11 +48,7 @@ public void destroy() {
     }
 
     @Override
-    public void flushPartition(Object partitionKey) {
-    }
-
-    @Override
-    public void flushPartitions(Predicate<Object> selector) {
+    public void flushPartitions(Predicate<Object> predicate) {
     }
 
     @Override

diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java
@@ -49,6 +49,7 @@
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 import javax.net.ssl.SSLEngine;
@@ -521,4 +522,8 @@ public ClientStats getClientStats() {
                 ));
         return new ClientStats(statsPerHost);
     }
+
+    public void flushChannelPoolPartitions(Predicate<Object> predicate) {
+        channelPool.flushPartitions(predicate);
+    }
 }
diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java b/client/src/main/java/org/asynchttpclient/netty/channel/DefaultChannelPool.java
@@ -353,16 +353,10 @@ private void flushPartition(Object partitionKey, ConcurrentLinkedDeque<IdleChann
     }
 
     @Override
-    public void flushPartition(Object partitionKey) {
-        flushPartition(partitionKey, partitions.get(partitionKey));
-    }
-
-    @Override
-    public void flushPartitions(Predicate<Object> selector) {
-
+    public void flushPartitions(Predicate<Object> predicate) {
         for (Map.Entry<Object, ConcurrentLinkedDeque<IdleChannel>> partitionsEntry : partitions.entrySet()) {
             Object partitionKey = partitionsEntry.getKey();
-            if (selector.test(partitionKey))
+            if (predicate.test(partitionKey))
                 flushPartition(partitionKey, partitionsEntry.getValue());
         }
     }

diff --git a/extras/registry/src/test/java/org/asynchttpclient/extras/registry/BadAsyncHttpClient.java b/extras/registry/src/test/java/org/asynchttpclient/extras/registry/BadAsyncHttpClient.java
@@ -12,6 +12,8 @@
  */
 package org.asynchttpclient.extras.registry;
 
+import java.util.function.Predicate;
+
 import org.asynchttpclient.AsyncHandler;
 import org.asynchttpclient.AsyncHttpClient;
 import org.asynchttpclient.BoundRequestBuilder;
@@ -131,4 +133,9 @@ public ListenableFuture<Response> executeRequest(RequestBuilder requestBuilder)
     public ClientStats getClientStats() {
         throw new UnsupportedOperationException();
     }
+
+    @Override
+    public void flushChannelPoolPartitions(Predicate<Object> predicate) {
+        throw new UnsupportedOperationException();
+    }
 }
diff --git a/extras/registry/src/test/java/org/asynchttpclient/extras/registry/TestAsyncHttpClient.java b/extras/registry/src/test/java/org/asynchttpclient/extras/registry/TestAsyncHttpClient.java
@@ -12,6 +12,8 @@
  */
 package org.asynchttpclient.extras.registry;
 
+import java.util.function.Predicate;
+
 import org.asynchttpclient.AsyncHandler;
 import org.asynchttpclient.AsyncHttpClient;
 import org.asynchttpclient.BoundRequestBuilder;
@@ -127,4 +129,9 @@ public ListenableFuture<Response> executeRequest(RequestBuilder requestBuilder)
     public ClientStats getClientStats() {
         throw new UnsupportedOperationException();
     }
+
+    @Override
+    public void flushChannelPoolPartitions(Predicate<Object> predicate) {
+        throw new UnsupportedOperationException();
+    }
 }