Add executor service argument to Channel#asyncCompletableRpc

RPC response comes from the reading threads and this can cause
deadlocks if the client doesn't use CompletableFuture#*async
methods.

References #215

Author: acogoluegnes

src/main/java/com/rabbitmq/client/Channel.java

@@ -18,6 +18,7 @@
 import java.io.IOException;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeoutException;
 
 import com.rabbitmq.client.AMQP.BasicProperties;
@@ -1367,9 +1368,10 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
     /**
      * Asynchronously send a method over this channel.
      * @param method method to transmit over this channel.
+     * @param executorService executor used to complete the operation, can be null
      * @return a completable future that completes when the result is received
      * @throws IOException Problem transmitting method.
      */
-    CompletableFuture<Command> asyncCompletableRpc(Method method) throws IOException;
+    CompletableFuture<Command> asyncCompletableRpc(Method method, ExecutorService executorService) throws IOException;
 
 }

src/main/java/com/rabbitmq/client/impl/AMQChannel.java

@@ -29,6 +29,7 @@
 
 import java.io.IOException;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Supplier;
 
@@ -145,11 +146,11 @@ public AMQCommand exnWrappingRpc(Method m)
         }
     }
 
-    public CompletableFuture<Command> exnWrappingAsyncRpc(Method m)
+    public CompletableFuture<Command> exnWrappingAsyncRpc(Method m, ExecutorService executorService)
         throws IOException
     {
         try {
-            return privateAsyncRpc(m);
+            return privateAsyncRpc(m, executorService);
         } catch (AlreadyClosedException ace) {
             // Do not wrap it since it means that connection/channel
             // was closed in some action in the past
@@ -204,8 +205,8 @@ public void enqueueRpc(RpcContinuation k)
         doEnqueueRpc(() -> new RpcContinuationRpcWrapper(k));
     }
 
-    public void enqueueAsyncRpc(Method method, CompletableFuture<Command> future) {
-        doEnqueueRpc(() -> new CompletableFutureRpcWrapper(method, future));
+    public void enqueueAsyncRpc(Method method, CompletableFuture<Command> future, ExecutorService executorService) {
+        doEnqueueRpc(() -> new CompletableFutureRpcWrapper(method, future, executorService));
     }
 
     private void doEnqueueRpc(Supplier<RpcWrapper> rpcWrapperSupplier) {
@@ -308,11 +309,11 @@ protected ChannelContinuationTimeoutException wrapTimeoutException(final Method
         return new ChannelContinuationTimeoutException(e, this, this._channelNumber, m);
     }
 
-    private CompletableFuture<Command> privateAsyncRpc(Method m)
+    private CompletableFuture<Command> privateAsyncRpc(Method m, ExecutorService executorService)
         throws IOException, ShutdownSignalException
     {
         CompletableFuture<Command> future = new CompletableFuture<>();
-        asyncRpc(m, future);
+        asyncRpc(m, future, executorService);
         return future;
     }
 
@@ -347,20 +348,20 @@ public void quiescingRpc(Method m, RpcContinuation k)
         }
     }
 
-    public void asyncRpc(Method m, CompletableFuture<Command> future)
+    public void asyncRpc(Method m, CompletableFuture<Command> future, ExecutorService executorService)
         throws IOException
     {
         synchronized (_channelMutex) {
             ensureIsOpen();
-            quiescingAsyncRpc(m, future);
+            quiescingAsyncRpc(m, future, executorService);
         }
     }
 
-    public void quiescingAsyncRpc(Method m, CompletableFuture<Command> future)
+    public void quiescingAsyncRpc(Method m, CompletableFuture<Command> future, ExecutorService executorService)
         throws IOException
     {
         synchronized (_channelMutex) {
-            enqueueAsyncRpc(m, future);
+            enqueueAsyncRpc(m, future, executorService);
             quiescingTransmit(m);
         }
     }

src/main/java/com/rabbitmq/client/impl/ChannelN.java

@@ -22,10 +22,7 @@
 import java.util.Map;
 import java.util.SortedSet;
 import java.util.TreeSet;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeoutException;
+import java.util.concurrent.*;
 
 import com.rabbitmq.client.ConfirmCallback;
 import com.rabbitmq.client.*;
@@ -1558,8 +1555,8 @@ public AMQCommand rpc(Method method) throws IOException {
     }
 
     @Override
-    public CompletableFuture<Command> asyncCompletableRpc(Method method) throws IOException {
-        return exnWrappingAsyncRpc(method);
+    public CompletableFuture<Command> asyncCompletableRpc(Method method, ExecutorService executorService) throws IOException {
+        return exnWrappingAsyncRpc(method, executorService);
     }
 
     @Override

src/main/java/com/rabbitmq/client/impl/CompletableFutureRpcWrapper.java

@@ -19,6 +19,8 @@
 import com.rabbitmq.client.Method;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 /**
  *
@@ -29,9 +31,12 @@ public class CompletableFutureRpcWrapper implements RpcWrapper {
 
     private final CompletableFuture<Command> completableFuture;
 
-    public CompletableFutureRpcWrapper(Method method, CompletableFuture<Command> completableFuture) {
+    private final ExecutorService executorService;
+
+    public CompletableFutureRpcWrapper(Method method, CompletableFuture<Command> completableFuture, ExecutorService executorService) {
         this.request = method;
         this.completableFuture = completableFuture;
+        this.executorService = executorService;
     }
 
     @Override
@@ -41,7 +46,11 @@ public boolean canHandleReply(AMQCommand command) {
 
     @Override
     public void complete(AMQCommand command) {
-        completableFuture.complete(command);
+        if (executorService == null) {
+            completableFuture.complete(command);
+        } else {
+            executorService.submit(() -> completableFuture.complete(command));
+        }
     }
 
     @Override

src/main/java/com/rabbitmq/client/impl/recovery/AutorecoveringChannel.java

@@ -22,6 +22,7 @@
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeoutException;
 
 /**
@@ -897,8 +898,8 @@ void updateConsumerTag(String tag, String newTag) {
     }
 
     @Override
-    public CompletableFuture<Command> asyncCompletableRpc(Method method) throws IOException {
-        return this.delegate.asyncCompletableRpc(method);
+    public CompletableFuture<Command> asyncCompletableRpc(Method method, ExecutorService executorService) throws IOException {
+        return this.delegate.asyncCompletableRpc(method, executorService);
     }
 
     @Override

src/test/java/com/rabbitmq/client/test/ChannelAsyncCompletableFutureTest.java

@@ -22,16 +22,37 @@
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
 import java.util.UUID;
-import java.util.concurrent.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.assertTrue;
 
+@RunWith(Parameterized.class)
 public class ChannelAsyncCompletableFutureTest extends BrokerTestCase {
 
-    ExecutorService executor;
+    ExecutorService executor, methodArgumentExecutor;
+
+    @Parameterized.Parameters
+    public static Collection<ExecutorService> params() {
+        List<ExecutorService> executors = new ArrayList<>();
+        executors.add(null);
+        executors.add(Executors.newSingleThreadExecutor());
+        return executors;
+    }
+
+    public ChannelAsyncCompletableFutureTest(ExecutorService methodArgumentExecutor) {
+        this.methodArgumentExecutor = methodArgumentExecutor;
+    }
 
     String queue;
     String exchange;
@@ -61,7 +82,7 @@ public void async() throws Exception {
             .arguments(null)
             .build();
 
-        channel.asyncCompletableRpc(queueDeclare)
+        channel.asyncCompletableRpc(queueDeclare, null)
             .thenComposeAsync(action -> {
                 try {
                     return channel.asyncCompletableRpc(new AMQImpl.Exchange.Declare.Builder()
@@ -70,7 +91,7 @@ public void async() throws Exception {
                         .durable(false)
                         .autoDelete(false)
                         .arguments(null)
-                        .build());
+                        .build(), methodArgumentExecutor);
                 } catch (IOException e) {
                     throw new RuntimeException(e);
                 }
@@ -81,7 +102,7 @@ public void async() throws Exception {
                         .exchange(exchange)
                         .routingKey("")
                         .arguments(null)
-                        .build());
+                        .build(), methodArgumentExecutor);
                 } catch (IOException e) {
                     throw new RuntimeException(e);
                 }