Use Executor type for dispatching, not ExecutorService

It allows other abstractions like Spring's TaskExecutor to be used.

References #160

Author: acogoluegnes

src/main/java/com/rabbitmq/client/amqp/ConnectionBuilder.java

@@ -17,6 +17,7 @@
 // [email protected].
 package com.rabbitmq.client.amqp;
 
+import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 
 /** Builder for {@link Connection} instances. */
@@ -38,21 +39,20 @@ public interface ConnectionBuilder extends ConnectionSettings<ConnectionBuilder>
   ConnectionBuilder listeners(Resource.StateListener... listeners);
 
   /**
-   * Set the executor service to use for incoming message delivery.
+   * Set the executor to use for incoming message delivery.
    *
-   * <p>The executor service is shared between the connection consumers.
+   * <p>The executor is shared between the connection consumers.
    *
-   * <p>By default, an executor service with {@link Runtime#availableProcessors()} thread(s) is
-   * created for the connection.
+   * <p>By default, an {@link ExecutorService} with {@link Runtime#availableProcessors()} thread(s)
+   * is created for the connection.
    *
    * <p>It is the developer's responsibility to shut down the executor when it is no longer needed.
    *
-   * @param executorService executor service for incoming message delivery
+   * @param executor executor for incoming message delivery
    * @return this builder instance
-   * @see
-   *     com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder#dispatchingExecutorService(ExecutorService)
+   * @see com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder#dispatchingExecutor(ExecutorService)
    */
-  ConnectionBuilder dispatchingExecutorService(ExecutorService executorService);
+  ConnectionBuilder dispatchingExecutor(Executor executor);
 
   /**
    * Create the connection instance.

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConnection.java

@@ -99,8 +99,8 @@ final class AmqpConnection extends ResourceBase implements Connection {
   private final Lock instanceLock = new ReentrantLock();
   private final boolean filterExpressionsSupported, setTokenSupported;
   private volatile ConsumerWorkService consumerWorkService;
-  private volatile ExecutorService dispatchingExecutorService;
-  private final boolean privateDispatchingExecutorService;
+  private volatile Executor dispatchingExecutor;
+  private final boolean privateDispatchingExecutor;
   private final CredentialsManager.Registration credentialsRegistration;
 
   AmqpConnection(AmqpConnectionBuilder builder) {
@@ -120,16 +120,16 @@ final class AmqpConnection extends ResourceBase implements Connection {
 
     this.topologyListener = createTopologyListener(builder);
 
-    ExecutorService des =
-        builder.dispatchingExecutorService() == null
+    Executor de =
+        builder.dispatchingExecutor() == null
             ? environment.dispatchingExecutorService()
-            : builder.dispatchingExecutorService();
+            : builder.dispatchingExecutor();
 
-    if (des == null) {
-      this.privateDispatchingExecutorService = true;
+    if (de == null) {
+      this.privateDispatchingExecutor = true;
     } else {
-      this.privateDispatchingExecutorService = false;
-      this.dispatchingExecutorService = des;
+      this.privateDispatchingExecutor = false;
+      this.dispatchingExecutor = de;
     }
 
     if (recoveryConfiguration.activated()) {
@@ -707,13 +707,13 @@ ConsumerWorkService consumerWorkService() {
     this.instanceLock.lock();
     try {
       if (this.consumerWorkService == null) {
-        if (this.privateDispatchingExecutorService) {
-          this.dispatchingExecutorService =
+        if (this.privateDispatchingExecutor) {
+          this.dispatchingExecutor =
               Executors.newFixedThreadPool(
                   DEFAULT_NUM_THREADS, Utils.threadFactory("dispatching-" + this.name() + "-"));
         }
         this.consumerWorkService =
-            new WorkPoolConsumerWorkService(this.dispatchingExecutorService, Duration.ZERO);
+            new WorkPoolConsumerWorkService(this.dispatchingExecutor, Duration.ZERO);
       }
       return this.consumerWorkService;
     } finally {
@@ -883,11 +883,11 @@ private void close(Throwable cause) {
         }
       }
       try {
-        if (this.privateDispatchingExecutorService) {
-          ExecutorService es = this.dispatchingExecutorService;
+        if (this.privateDispatchingExecutor) {
+          Executor es = this.dispatchingExecutor;
           if (es != null) {
             try {
-              es.shutdownNow();
+              ((ExecutorService) es).shutdownNow();
             } catch (Exception e) {
               LOGGER.info(
                   "Error while shutting down dispatching executor service for connection '{}': {}",

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConnectionBuilder.java

@@ -22,7 +22,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executor;
 
 class AmqpConnectionBuilder implements ConnectionBuilder {
 
@@ -32,7 +32,7 @@ class AmqpConnectionBuilder implements ConnectionBuilder {
   private final DefaultConnectionSettings<AmqpConnectionBuilder> connectionSettings =
       new AmqpConnectionBuilderConnectionSettings(this);
   private final List<Resource.StateListener> listeners = new ArrayList<>();
-  private ExecutorService dispatchingExecutorService;
+  private Executor dispatchingExecutor;
   private String name;
   private TopologyListener topologyListener;
   private boolean isolateResources = false;
@@ -123,8 +123,8 @@ public ConnectionBuilder listeners(Resource.StateListener... listeners) {
   }
 
   @Override
-  public ConnectionBuilder dispatchingExecutorService(ExecutorService executorService) {
-    this.dispatchingExecutorService = executorService;
+  public ConnectionBuilder dispatchingExecutor(Executor executor) {
+    this.dispatchingExecutor = executor;
     return this;
   }
 
@@ -143,8 +143,8 @@ boolean isolateResources() {
     return isolateResources;
   }
 
-  ExecutorService dispatchingExecutorService() {
-    return this.dispatchingExecutorService;
+  Executor dispatchingExecutor() {
+    return this.dispatchingExecutor;
   }
 
   @Override
@@ -159,7 +159,7 @@ void copyTo(AmqpConnectionBuilder copy) {
     copy.name(this.name);
     copy.topologyListener(this.topologyListener);
     copy.isolateResources(this.isolateResources);
-    copy.dispatchingExecutorService(this.dispatchingExecutorService);
+    copy.dispatchingExecutor(this.dispatchingExecutor);
   }
 
   AmqpConnectionBuilder name(String name) {

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConsumer.java

@@ -314,9 +314,11 @@ void close(Throwable cause) {
       } catch (Exception e) {
         LOGGER.warn("Error while closing receiver", e);
       }
-
       this.state(CLOSED, cause);
-      this.metricsCollector.closeConsumer();
+      MetricsCollector mc = this.metricsCollector;
+      if (mc != null) {
+        mc.closeConsumer();
+      }
     }
   }
 

src/main/java/com/rabbitmq/client/amqp/impl/AmqpEnvironment.java

@@ -43,7 +43,7 @@ class AmqpEnvironment implements Environment {
   private final boolean internalPublisherExecutor;
   private final ExecutorService executorService;
   private final ScheduledExecutorService scheduledExecutorService;
-  private final ExecutorService dispatchingExecutorService;
+  private final Executor dispatchingExecutorService;
   private final ExecutorService publisherExecutorService;
   private final ConnectionManager connectionManager = new ConnectionManager(this);
   private final long id;
@@ -61,7 +61,7 @@ class AmqpEnvironment implements Environment {
   AmqpEnvironment(
       ExecutorService executorService,
       ScheduledExecutorService scheduledExecutorService,
-      ExecutorService dispatchingExecutorService,
+      Executor dispatchingExecutorService,
       ExecutorService publisherExecutorService,
       DefaultConnectionSettings<?> connectionSettings,
       MetricsCollector metricsCollector,
@@ -166,7 +166,7 @@ ExecutorService executorService() {
     return this.executorService;
   }
 
-  ExecutorService dispatchingExecutorService() {
+  Executor dispatchingExecutorService() {
     return this.dispatchingExecutorService;
   }
 

src/main/java/com/rabbitmq/client/amqp/impl/AmqpEnvironmentBuilder.java

@@ -21,6 +21,7 @@
 import com.rabbitmq.client.amqp.metrics.MetricsCollector;
 import com.rabbitmq.client.amqp.metrics.NoOpMetricsCollector;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
 
@@ -31,7 +32,7 @@ public class AmqpEnvironmentBuilder implements EnvironmentBuilder {
       new DefaultEnvironmentConnectionSettings(this);
   private ExecutorService executorService;
   private ScheduledExecutorService scheduledExecutorService;
-  private ExecutorService dispatchingExecutorService;
+  private Executor dispatchingExecutor;
   private ExecutorService publisherExecutorService;
   private MetricsCollector metricsCollector = NoOpMetricsCollector.INSTANCE;
   private ObservationCollector observationCollector = Utils.NO_OP_OBSERVATION_COLLECTOR;
@@ -52,20 +53,20 @@ public AmqpEnvironmentBuilder executorService(ExecutorService executorService) {
   }
 
   /**
-   * Set the shared executor service to use for incoming message delivery in this environment
-   * instance connections.
+   * Set the shared executor to use for incoming message delivery in this environment instance
+   * connections.
    *
-   * <p>There is no shared executor service by default, each connection uses its own, see {@link
-   * ConnectionBuilder#dispatchingExecutorService(ExecutorService)}.
+   * <p>There is no shared executor by default, each connection uses its own, see {@link
+   * ConnectionBuilder#dispatchingExecutor(Executor)}.
    *
    * <p>It is the developer's responsibility to shut down the executor when it is no longer needed.
    *
-   * @param executorService the executor service for incoming message delivery
+   * @param executor the executor for incoming message delivery
    * @return this builder instance
-   * @see ConnectionBuilder#dispatchingExecutorService(ExecutorService)
+   * @see ConnectionBuilder#dispatchingExecutor(Executor)
    */
-  public AmqpEnvironmentBuilder dispatchingExecutorService(ExecutorService executorService) {
-    this.dispatchingExecutorService = executorService;
+  public AmqpEnvironmentBuilder dispatchingExecutor(Executor executor) {
+    this.dispatchingExecutor = executor;
     return this;
   }
 
@@ -161,7 +162,7 @@ public Environment build() {
     return new AmqpEnvironment(
         executorService,
         scheduledExecutorService,
-        dispatchingExecutorService,
+        dispatchingExecutor,
         publisherExecutorService,
         connectionSettings,
         metricsCollector,

src/main/java/com/rabbitmq/client/amqp/impl/AmqpPublisher.java

@@ -216,7 +216,10 @@ void close(Throwable cause) {
           this.sessionHandler,
           e -> LOGGER.info("Error while closing publisher session handler", e));
       this.state(State.CLOSED, cause);
-      this.metricsCollector.closePublisher();
+      MetricsCollector mc = this.metricsCollector;
+      if (mc != null) {
+        mc.closePublisher();
+      }
     }
   }
 

src/main/java/com/rabbitmq/client/amqp/impl/WorkPoolConsumerWorkService.java

@@ -20,16 +20,16 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executor;
 
 final class WorkPoolConsumerWorkService implements ConsumerWorkService {
 
   private static final int MAX_RUNNABLE_BLOCK_SIZE = 256;
 
-  private final ExecutorService executor;
+  private final Executor executor;
   private final WorkPool<AmqpConsumer, Runnable> workPool;
 
-  WorkPoolConsumerWorkService(ExecutorService executorService, Duration queueingTimeout) {
+  WorkPoolConsumerWorkService(Executor executorService, Duration queueingTimeout) {
     this.executor = executorService;
     this.workPool = new WorkPool<>(queueingTimeout);
   }

src/test/java/com/rabbitmq/client/amqp/impl/AmqpTest.java

@@ -801,8 +801,7 @@ void messagesAreDispatchedToEnvironmentConnectionExecutorService() {
     String connPrefix = "conn-";
     ExecutorService envExecutor = newSingleThreadExecutor(threadFactory(envPrefix));
     ExecutorService connExecutor = newSingleThreadExecutor(threadFactory(connPrefix));
-    Environment env =
-        TestUtils.environmentBuilder().dispatchingExecutorService(envExecutor).build();
+    Environment env = TestUtils.environmentBuilder().dispatchingExecutor(envExecutor).build();
     try {
       BiConsumer<Connection, String> operation =
           (c, prefix) -> {
@@ -825,7 +824,7 @@ void messagesAreDispatchedToEnvironmentConnectionExecutorService() {
           };
 
       Connection c1 = env.connectionBuilder().build();
-      Connection c2 = env.connectionBuilder().dispatchingExecutorService(connExecutor).build();
+      Connection c2 = env.connectionBuilder().dispatchingExecutor(connExecutor).build();
 
       operation.accept(c1, envPrefix);
       operation.accept(c2, connPrefix);

src/test/java/com/rabbitmq/client/amqp/impl/Oauth2Test.java

@@ -230,6 +230,7 @@ void tokenShouldBeRefreshedAutomatically(boolean shared, TestInfo info) throws E
   }
 
   @Test
+  @BrokerVersionAtLeast(RABBITMQ_4_1_0)
   void tokenOnHttpsShouldBeRefreshed(TestInfo info) throws Exception {
     KeyStore keyStore = generateKeyPair();
 

src/test/java/com/rabbitmq/client/amqp/impl/RecoveryClusterTest.java

@@ -82,7 +82,7 @@ void init(TestInfo info) {
         Executors.newSingleThreadExecutor(Utils.threadFactory("env-dispatching-executor-service-"));
     environment =
         new AmqpEnvironmentBuilder()
-            .dispatchingExecutorService(dispatchingExecutorService)
+            .dispatchingExecutor(dispatchingExecutorService)
             .connectionSettings()
             .uris(URIS)
             .environmentBuilder()