code review comments

Author: vikinghawk

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

@@ -127,7 +127,7 @@ public class ConnectionFactory implements Cloneable {
 
     private boolean automaticRecovery               = true;
     private boolean topologyRecovery                = true;
-    private int topologyRecoveryThreads             = 1;
+    private ExecutorService topologyRecoveryExecutor;
 
     // long is used to make sure the users can use both ints
     // and longs safely. It is unlikely that anybody'd need
@@ -714,13 +714,22 @@ public void setTopologyRecoveryEnabled(boolean topologyRecovery) {
         this.topologyRecovery = topologyRecovery;
     }
 
-    public int getTopologyRecoveryThreadCount() {
-        return topologyRecoveryThreads;
+    /**
+     * Get the executor to use for parallel topology recovery. If null (the default), recovery is done single threaded on the main connection thread.
+     * @return thread pool executor
+     */
+    public ExecutorService getTopologyRecoveryExecutor() {
+        return topologyRecoveryExecutor;
     }
 
-    // TODO Document that your exception handler method should be thread safe
-    public void setTopologyRecoveryThreadCount(final int topologyRecoveryThreads) {
-        this.topologyRecoveryThreads = topologyRecoveryThreads;
+    /**
+     * Set the executor to use for parallel topology recovery. If null (the default), recovery is done single threaded on the main connection thread.
+     * It is recommended to pass a ThreadPoolExecutor that will allow its core threads to timeout so these threads can die when recovery is complete.
+     * Note: your {@link ExceptionHandler#handleTopologyRecoveryException(Connection, Channel, TopologyRecoveryException)} method should be thread-safe.
+     * @param topologyRecoveryExecutor thread pool executor
+     */
+    public void setTopologyRecoveryExecutor(final ExecutorService topologyRecoveryExecutor) {
+        this.topologyRecoveryExecutor = topologyRecoveryExecutor;
     }
 
     public void setMetricsCollector(MetricsCollector metricsCollector) {
@@ -1021,7 +1030,7 @@ public ConnectionParams params(ExecutorService consumerWorkServiceExecutor) {
         result.setNetworkRecoveryInterval(networkRecoveryInterval);
         result.setRecoveryDelayHandler(recoveryDelayHandler);
         result.setTopologyRecovery(topologyRecovery);
-        result.setTopologyRecoveryThreadCount(topologyRecoveryThreads);
+        result.setTopologyRecoveryExecutor(topologyRecoveryExecutor);
         result.setExceptionHandler(exceptionHandler);
         result.setThreadFactory(threadFactory);
         result.setHandshakeTimeout(handshakeTimeout);

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

@@ -108,7 +108,7 @@ void handleConsumerException(Channel channel,
      * during topology (exchanges, queues, bindings, consumers) recovery
      * that it can't otherwise deal with.
      * @param conn the Connection that caught the exception
-     * @param ch the Channel that caught the exception
+     * @param ch the Channel that caught the exception. May be null.
      * @param exception the exception caught in the driver thread
      */
 

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

@@ -41,7 +41,7 @@ public class ConnectionParams {
     private long networkRecoveryInterval;
     private RecoveryDelayHandler recoveryDelayHandler;
     private boolean topologyRecovery;
-    private int topologyRecoveryThreads = 1;
+    private ExecutorService topologyRecoveryExecutor;
     private int channelRpcTimeout;
     private boolean channelShouldCheckRpcResponseType;
     private ErrorOnWriteListener errorOnWriteListener;
@@ -116,8 +116,12 @@ public boolean isTopologyRecoveryEnabled() {
         return topologyRecovery;
     }
 
-    public int getTopologyRecoveryThreadCount() {
-        return topologyRecoveryThreads;
+    /**
+     * Get the topology recovery executor. If null, the main connection thread should be used.
+     * @return executor. May be null.
+     */
+    public ExecutorService getTopologyRecoveryExecutor() {
+        return topologyRecoveryExecutor;
     }
 
     public ThreadFactory getThreadFactory() {
@@ -180,8 +184,8 @@ public void setTopologyRecovery(boolean topologyRecovery) {
         this.topologyRecovery = topologyRecovery;
     }
 
-    public void setTopologyRecoveryThreadCount(final int topologyRecoveryThreads) {
-        this.topologyRecoveryThreads = topologyRecoveryThreads;
+    public void setTopologyRecoveryExecutor(final ExecutorService topologyRecoveryExecutor) {
+        this.topologyRecoveryExecutor = topologyRecoveryExecutor;
     }
 
     public void setExceptionHandler(ExceptionHandler exceptionHandler) {

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

@@ -544,7 +544,7 @@ private synchronized void beginAutomaticRecovery() throws InterruptedException {
 	    // don't assign new delegate connection until channel recovery is complete
 	    this.delegate = newConn;
 	    if (this.params.isTopologyRecoveryEnabled()) {
-	        recoverTopology(params.getTopologyRecoveryThreadCount());
+	        recoverTopology(params.getTopologyRecoveryExecutor());
 	    }
 		this.notifyRecoveryListenersComplete();
     }
@@ -612,27 +612,13 @@ private void notifyRecoveryListenersStarted() {
         }
     }
 
-    private void recoverTopology(final int recoveryThreads) throws InterruptedException {
+    private void recoverTopology(final ExecutorService executor) {
         // The recovery sequence is the following:
         // 1. Recover exchanges
         // 2. Recover queues
         // 3. Recover bindings
         // 4. Recover consumers
-        if (recoveryThreads > 1) {
-            // Support recovering entities in parallel for connections that have a lot of queues, bindings, & consumers
-            // A channel is single threaded, so group things by channel and recover 1 entity at a time per channel
-            // We still need to recover 1 type of entity at a time in case channel1 has a binding to a queue that is currently owned and being recovered by channel2 for example
-            final ExecutorService executor = Executors.newFixedThreadPool(recoveryThreads, delegate.getThreadFactory());
-            try {
-                // invokeAll will block until all callables are completed
-                executor.invokeAll(groupEntitiesByChannel(Utility.copy(recordedExchanges).values()));
-                executor.invokeAll(groupEntitiesByChannel(Utility.copy(recordedQueues).values()));
-                executor.invokeAll(groupEntitiesByChannel(Utility.copy(recordedBindings)));
-                executor.invokeAll(groupEntitiesByChannel(Utility.copy(consumers).values()));
-            } finally {
-                executor.shutdownNow();
-            }
-        } else {
+        if (executor == null) {
             // recover entities in serial on the main connection thread
             for (final RecordedExchange exchange : Utility.copy(recordedExchanges).values()) {
                 recoverExchange(exchange);
@@ -646,6 +632,21 @@ private void recoverTopology(final int recoveryThreads) throws InterruptedExcept
             for (final Map.Entry<String, RecordedConsumer> entry : Utility.copy(consumers).entrySet()) {
                 recoverConsumer(entry.getKey(), entry.getValue());
             }
+        } else {
+            // Support recovering entities in parallel for connections that have a lot of queues, bindings, & consumers
+            // A channel is single threaded, so group things by channel and recover 1 entity at a time per channel
+            // We also need to recover 1 type of entity at a time in case channel1 has a binding to a queue that is currently owned and being recovered by channel2 for example
+            // Note: invokeAll will block until all callables are completed and all returned futures will be complete 
+            try {
+                executor.invokeAll(groupEntitiesByChannel(Utility.copy(recordedExchanges).values()));
+                executor.invokeAll(groupEntitiesByChannel(Utility.copy(recordedQueues).values()));
+                executor.invokeAll(groupEntitiesByChannel(Utility.copy(recordedBindings)));
+                executor.invokeAll(groupEntitiesByChannel(Utility.copy(consumers).values()));
+            } catch (final Exception cause) {
+                final String message = "Caught an exception while recovering toplogy: " + cause.getMessage();
+                final TopologyRecoveryException e = new TopologyRecoveryException(message, cause);
+                getExceptionHandler().handleTopologyRecoveryException(delegate, null, e);
+            }
         }
     }
 

src/test/java/com/rabbitmq/client/test/functional/ConnectionRecovery.java

@@ -32,6 +32,9 @@
 
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -809,8 +812,12 @@ public void handleDelivery(String consumerTag,
 
     @Test public void recoveryWithMultipleThreads() throws Exception {
         // test with 8 recovery threads
-        ConnectionFactory connectionFactory = buildConnectionFactoryWithRecoveryEnabled(false, 8);
-        assertEquals(8, connectionFactory.getTopologyRecoveryThreadCount());
+        final ThreadPoolExecutor executor = new ThreadPoolExecutor(8, 8, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
+        executor.allowCoreThreadTimeOut(true);
+        ConnectionFactory connectionFactory = buildConnectionFactoryWithRecoveryEnabled(false);
+        assertNull(connectionFactory.getTopologyRecoveryExecutor());
+        connectionFactory.setTopologyRecoveryExecutor(executor);
+        assertEquals(executor, connectionFactory.getTopologyRecoveryExecutor());
         RecoverableConnection testConnection = (RecoverableConnection) connectionFactory.newConnection();
         try {
             final List<Channel> channels = new ArrayList<Channel>();
@@ -822,6 +829,7 @@ public void handleDelivery(String consumerTag,
             final CountDownLatch latch = new CountDownLatch(channelCount * queuesPerChannel);
             for (int i=0; i < channelCount; i++) {
                 final Channel testChannel = testConnection.createChannel();
+                channels.add(testChannel);
                 String x = "tmp-x-topic-" + i;
                 exchanges.add(x);
                 testChannel.exchangeDeclare(x, "topic");
@@ -1019,17 +1027,12 @@ private static RecoverableConnection newRecoveringConnection(String connectionNa
     }
 
     private static ConnectionFactory buildConnectionFactoryWithRecoveryEnabled(boolean disableTopologyRecovery) {
-        return buildConnectionFactoryWithRecoveryEnabled(disableTopologyRecovery, 1);
-    }
-
-    private static ConnectionFactory buildConnectionFactoryWithRecoveryEnabled(boolean disableTopologyRecovery, final int recoveryThreads) {
         ConnectionFactory cf = TestUtils.connectionFactory();
         cf.setNetworkRecoveryInterval(RECOVERY_INTERVAL);
         cf.setAutomaticRecoveryEnabled(true);
         if (disableTopologyRecovery) {
             cf.setTopologyRecoveryEnabled(false);
         }
-        cf.setTopologyRecoveryThreadCount(recoveryThreads);
         return cf;
     }