Remove executor service argument from Channel#asyncCompletableRpc

Even if CompletableFuture is completed in the reading thread,
client API (e.g. CompletableStage#then*, Reactor Flux/Mono)
provide ways to control the threading behavior.

References #215

Author: acogoluegnes

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

@@ -18,7 +18,6 @@
 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;
@@ -1368,10 +1367,9 @@ 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, ExecutorService executorService) throws IOException;
+    CompletableFuture<Command> asyncCompletableRpc(Method method) throws IOException;
 
 }

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

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

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

@@ -1555,8 +1555,8 @@ public AMQCommand rpc(Method method) throws IOException {
     }
 
     @Override
-    public CompletableFuture<Command> asyncCompletableRpc(Method method, ExecutorService executorService) throws IOException {
-        return exnWrappingAsyncRpc(method, executorService);
+    public CompletableFuture<Command> asyncCompletableRpc(Method method) throws IOException {
+        return exnWrappingAsyncRpc(method);
     }
 
     @Override

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

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

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

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

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

@@ -22,13 +22,8 @@
 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.CountDownLatch;
 import java.util.concurrent.ExecutorService;
@@ -37,22 +32,9 @@
 
 import static org.junit.Assert.assertTrue;
 
-@RunWith(Parameterized.class)
 public class ChannelAsyncCompletableFutureTest extends BrokerTestCase {
 
-    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;
-    }
+    ExecutorService executor;
 
     String queue;
     String exchange;
@@ -67,9 +49,6 @@ public ChannelAsyncCompletableFutureTest(ExecutorService methodArgumentExecutor)
         executor.shutdownNow();
         channel.queueDelete(queue);
         channel.exchangeDelete(exchange);
-        if (methodArgumentExecutor != null) {
-            methodArgumentExecutor.shutdownNow();
-        }
     }
 
     @Test
@@ -85,7 +64,7 @@ public void async() throws Exception {
             .arguments(null)
             .build();
 
-        channel.asyncCompletableRpc(queueDeclare, null)
+        channel.asyncCompletableRpc(queueDeclare)
             .thenComposeAsync(action -> {
                 try {
                     return channel.asyncCompletableRpc(new AMQImpl.Exchange.Declare.Builder()
@@ -94,7 +73,7 @@ public void async() throws Exception {
                         .durable(false)
                         .autoDelete(false)
                         .arguments(null)
-                        .build(), methodArgumentExecutor);
+                        .build());
                 } catch (IOException e) {
                     throw new RuntimeException(e);
                 }
@@ -105,7 +84,7 @@ public void async() throws Exception {
                         .exchange(exchange)
                         .routingKey("")
                         .arguments(null)
-                        .build(), methodArgumentExecutor);
+                        .build());
                 } catch (IOException e) {
                     throw new RuntimeException(e);
                 }