Add dispatching executor service setting to connection and consumer

References #160

Author: acogoluegnes

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

@@ -17,6 +17,8 @@
 // [email protected].
 package com.rabbitmq.client.amqp;
 
+import java.util.concurrent.ExecutorService;
+
 /** Builder for {@link Connection} instances. */
 public interface ConnectionBuilder extends ConnectionSettings<ConnectionBuilder> {
 
@@ -35,6 +37,22 @@ public interface ConnectionBuilder extends ConnectionSettings<ConnectionBuilder>
    */
   ConnectionBuilder listeners(Resource.StateListener... listeners);
 
+  /**
+   * Set the executor service to use for incoming message delivery.
+   *
+   * <p>The executor service is shared between the connection consumers, unless a consumer sets its
+   * own executor service with {@link ConsumerBuilder#dispatchingExecutorService(ExecutorService)}.
+   *
+   * <p>By default, a new single-threaded executor is created.
+   *
+   * <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
+   * @return this builder instance
+   * @see ConsumerBuilder#dispatchingExecutorService(ExecutorService)
+   */
+  ConnectionBuilder dispatchingExecutorService(ExecutorService executorService);
+
   /**
    * Create the connection instance.
    *

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

@@ -20,6 +20,7 @@
 import java.math.BigDecimal;
 import java.time.Instant;
 import java.util.UUID;
+import java.util.concurrent.ExecutorService;
 import org.apache.qpid.protonj2.types.*;
 
 /** API to configure and create a {@link Consumer}. */
@@ -68,6 +69,19 @@ public interface ConsumerBuilder {
    */
   ConsumerBuilder listeners(Resource.StateListener... listeners);
 
+  /**
+   * Set the executor service to use for incoming message delivery for this consumer.
+   *
+   * <p>The consumer uses the connection's executor service by default.
+   *
+   * <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
+   * @return this builder instance
+   * @see ConnectionBuilder#dispatchingExecutorService(ExecutorService)
+   */
+  ConsumerBuilder dispatchingExecutorService(ExecutorService executorService);
+
   /**
    * Options for a consumer consuming from a stream.
    *

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

@@ -97,6 +97,7 @@ final class AmqpConnection extends ResourceBase implements Connection {
   private final Lock instanceLock = new ReentrantLock();
   private final boolean filterExpressionsSupported, setTokenSupported;
   private volatile ExecutorService dispatchingExecutorService;
+  private final boolean privateDispatchingExecutorService;
   private final CredentialsManager.Registration credentialsRegistration;
 
   AmqpConnection(AmqpConnectionBuilder builder) {
@@ -116,6 +117,13 @@ final class AmqpConnection extends ResourceBase implements Connection {
 
     this.topologyListener = createTopologyListener(builder);
 
+    if (builder.dispatchingExecutorService() == null) {
+      this.privateDispatchingExecutorService = true;
+    } else {
+      this.privateDispatchingExecutorService = false;
+      this.dispatchingExecutorService = builder.dispatchingExecutorService();
+    }
+
     if (recoveryConfiguration.activated()) {
       disconnectHandler = recoveryDisconnectHandler(recoveryConfiguration, this.name());
     } else {
@@ -853,15 +861,17 @@ private void close(Throwable cause) {
         LOGGER.info("Interrupted while waiting for connection lock");
       }
       try {
-        ExecutorService es = this.dispatchingExecutorService;
-        if (es != null) {
-          try {
-            es.shutdownNow();
-          } catch (Exception e) {
-            LOGGER.info(
-                "Error while shutting down dispatching executor service for connection '{}': {}",
-                this.name(),
-                e.getMessage());
+        if (this.privateDispatchingExecutorService) {
+          ExecutorService es = this.dispatchingExecutorService;
+          if (es != null) {
+            try {
+              es.shutdownNow();
+            } catch (Exception e) {
+              LOGGER.info(
+                  "Error while shutting down dispatching executor service for connection '{}': {}",
+                  this.name(),
+                  e.getMessage());
+            }
           }
         }
         try {

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

@@ -22,6 +22,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
 
 class AmqpConnectionBuilder implements ConnectionBuilder {
 
@@ -31,6 +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 String name;
   private TopologyListener topologyListener;
   private boolean isolateResources = false;
@@ -120,6 +122,12 @@ public ConnectionBuilder listeners(Resource.StateListener... listeners) {
     return this;
   }
 
+  @Override
+  public ConnectionBuilder dispatchingExecutorService(ExecutorService executorService) {
+    this.dispatchingExecutorService = executorService;
+    return this;
+  }
+
   @Override
   public RecoveryConfiguration recovery() {
     this.recoveryConfiguration.activated(true);
@@ -135,6 +143,10 @@ boolean isolateResources() {
     return isolateResources;
   }
 
+  ExecutorService dispatchingExecutorService() {
+    return this.dispatchingExecutorService;
+  }
+
   @Override
   public Connection build() {
     return this.environment.connection(this);
@@ -147,6 +159,7 @@ void copyTo(AmqpConnectionBuilder copy) {
     copy.name(this.name);
     copy.topologyListener(this.topologyListener);
     copy.isolateResources(this.isolateResources);
+    copy.dispatchingExecutorService(this.dispatchingExecutorService);
   }
 
   AmqpConnectionBuilder name(String name) {

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

@@ -105,7 +105,10 @@ final class AmqpConsumer extends ResourceBase implements Consumer {
     this.connection = builder.connection();
     this.sessionHandler = this.connection.createSessionHandler();
 
-    this.dispatchingExecutorService = connection.dispatchingExecutorService();
+    this.dispatchingExecutorService =
+        builder.dispatchingExecutorService() == null
+            ? connection.dispatchingExecutorService()
+            : builder.dispatchingExecutorService();
     this.nativeHandler = createNativeHandler(messageHandler);
     this.nativeCloseHandler =
         e ->

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

@@ -25,6 +25,7 @@
 import java.math.BigDecimal;
 import java.time.Instant;
 import java.util.*;
+import java.util.concurrent.ExecutorService;
 import org.apache.qpid.protonj2.types.*;
 
 class AmqpConsumerBuilder implements ConsumerBuilder {
@@ -36,6 +37,7 @@ class AmqpConsumerBuilder implements ConsumerBuilder {
   private Consumer.MessageHandler messageHandler;
   private int initialCredits = 100;
   private final List<Resource.StateListener> listeners = new ArrayList<>();
+  private ExecutorService dispatchingExecutorService;
   private final Map<String, DescribedType> filters = new LinkedHashMap<>();
   private final Map<String, Object> properties = new LinkedHashMap<>();
   private final StreamOptions streamOptions = new DefaultStreamOptions(this, this.filters);
@@ -79,6 +81,12 @@ public ConsumerBuilder listeners(Resource.StateListener... listeners) {
     return this;
   }
 
+  @Override
+  public ConsumerBuilder dispatchingExecutorService(ExecutorService executorService) {
+    this.dispatchingExecutorService = executorService;
+    return this;
+  }
+
   @Override
   public StreamOptions stream() {
     return this.streamOptions;
@@ -118,6 +126,10 @@ List<Resource.StateListener> listeners() {
     return listeners;
   }
 
+  ExecutorService dispatchingExecutorService() {
+    return this.dispatchingExecutorService;
+  }
+
   Map<String, DescribedType> filters() {
     return this.filters;
   }