recover channels in parallel too

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 int recoveryThreads                     = 1;
 
     // 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,13 @@ public void setTopologyRecoveryEnabled(boolean topologyRecovery) {
         this.topologyRecovery = topologyRecovery;
     }
 
-    public int getTopologyRecoveryThreadCount() {
-        return topologyRecoveryThreads;
+    public int getRecoveryThreadCount() {
+        return recoveryThreads;
     }
 
     // TODO Document that your exception handler method should be thread safe
-    public void setTopologyRecoveryThreadCount(final int topologyRecoveryThreads) {
-        this.topologyRecoveryThreads = topologyRecoveryThreads;
+    public void setRecoveryThreadCount(final int recoveryThreads) {
+        this.recoveryThreads = recoveryThreads;
     }
 
     public void setMetricsCollector(MetricsCollector metricsCollector) {
@@ -1021,7 +1021,7 @@ public ConnectionParams params(ExecutorService consumerWorkServiceExecutor) {
         result.setNetworkRecoveryInterval(networkRecoveryInterval);
         result.setRecoveryDelayHandler(recoveryDelayHandler);
         result.setTopologyRecovery(topologyRecovery);
-        result.setTopologyRecoveryThreadCount(topologyRecoveryThreads);
+        result.setRecoveryThreadCount(recoveryThreads);
         result.setExceptionHandler(exceptionHandler);
         result.setThreadFactory(threadFactory);
         result.setHandshakeTimeout(handshakeTimeout);

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 int recoveryThreads = 1;
     private int channelRpcTimeout;
     private boolean channelShouldCheckRpcResponseType;
     private ErrorOnWriteListener errorOnWriteListener;
@@ -116,8 +116,8 @@ public boolean isTopologyRecoveryEnabled() {
         return topologyRecovery;
     }
 
-    public int getTopologyRecoveryThreadCount() {
-        return topologyRecoveryThreads;
+    public int getRecoveryThreadCount() {
+        return recoveryThreads;
     }
 
     public ThreadFactory getThreadFactory() {
@@ -180,8 +180,8 @@ public void setTopologyRecovery(boolean topologyRecovery) {
         this.topologyRecovery = topologyRecovery;
     }
 
-    public void setTopologyRecoveryThreadCount(final int topologyRecoveryThreads) {
-        this.topologyRecoveryThreads = topologyRecoveryThreads;
+    public void setRecoveryThreadCount(final int recoveryThreads) {
+        this.recoveryThreads = recoveryThreads;
     }
 
     public void setExceptionHandler(ExceptionHandler exceptionHandler) {

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

@@ -540,12 +540,25 @@ private synchronized void beginAutomaticRecovery() throws InterruptedException {
         this.addAutomaticRecoveryListener(newConn);
 	    this.recoverShutdownListeners(newConn);
 	    this.recoverBlockedListeners(newConn);
-	    this.recoverChannels(newConn);
-	    // don't assign new delegate connection until channel recovery is complete
-	    this.delegate = newConn;
-	    if (this.params.isTopologyRecoveryEnabled()) {
-	        recoverTopology(params.getTopologyRecoveryThreadCount());
-	    }
+	    
+	    // Optionally support recovering channels & entities in parallel for connections that have a lot of channels, queues, bindings, etc.
+	    ExecutorService executor = null;
+        if (params.getRecoveryThreadCount() > 1) {
+            executor = Executors.newFixedThreadPool(params.getRecoveryThreadCount(), delegate.getThreadFactory());
+        }
+        try {
+    	    this.recoverChannels(newConn, executor);
+    	    // don't assign new delegate connection until channel recovery is complete
+    	    this.delegate = newConn;
+    	    // recover topology
+    	    if (this.params.isTopologyRecoveryEnabled()) {
+    	        recoverTopology(executor);
+    	    }
+        } finally {
+            if (executor != null) {
+                executor.shutdownNow();
+            }
+        }
 		this.notifyRecoveryListenersComplete();
     }
 
@@ -589,16 +602,34 @@ private RecoveryAwareAMQConnection recoverConnection() throws InterruptedExcepti
 		return null;
     }
 
-    private void recoverChannels(final RecoveryAwareAMQConnection newConn) {
-        for (AutorecoveringChannel ch : this.channels.values()) {
-            try {
-                ch.automaticallyRecover(this, newConn);
-                LOGGER.debug("Channel {} has recovered", ch);
-            } catch (Throwable t) {
-                newConn.getExceptionHandler().handleChannelRecoveryException(ch, t);
+    private void recoverChannels(final RecoveryAwareAMQConnection newConn, final ExecutorService executor) throws InterruptedException {
+        if (executor != null) {
+            final List<Callable<Object>> tasks = new ArrayList<Callable<Object>>();
+            for (final AutorecoveringChannel ch : this.channels.values()) {
+                tasks.add(Executors.callable(new Runnable() {
+                    @Override
+                    public void run() {
+                        recoverChannel(newConn, ch);
+                    }
+                }));
+            }
+            // invokeAll will block until all callables are completed
+            executor.invokeAll(tasks);
+        } else {
+            for (final AutorecoveringChannel ch : this.channels.values()) {
+                recoverChannel(newConn, ch);
             }
         }
     }
+    
+    private void recoverChannel(final RecoveryAwareAMQConnection newConn, final AutorecoveringChannel ch) {
+        try {
+            ch.automaticallyRecover(this, newConn);
+            LOGGER.debug("Channel {} has recovered", ch);
+        } catch (Throwable t) {
+            newConn.getExceptionHandler().handleChannelRecoveryException(ch, t);
+        }
+    }
 
     private void notifyRecoveryListenersComplete() {
         for (RecoveryListener f : Utility.copy(this.recoveryListeners)) {
@@ -612,26 +643,20 @@ private void notifyRecoveryListenersStarted() {
         }
     }
 
-    private void recoverTopology(final int recoveryThreads) throws InterruptedException {
+    private void recoverTopology(final ExecutorService executor) throws InterruptedException {
         // 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
+        if (executor != null) {
             // 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();
-            }
+            // 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()));
         } else {
             // recover entities in serial on the main connection thread
             for (final RecordedExchange exchange : Utility.copy(recordedExchanges).values()) {
@@ -760,9 +785,9 @@ private <E extends RecordedEntity> List<Callable<Object>> groupEntitiesByChannel
             list.add(entity);
         }
         // now create a runnable per channel
-        final List<Callable<Object>> callables = new ArrayList<Callable<Object>>();
+        final List<Callable<Object>> tasks = new ArrayList<Callable<Object>>();
         for (final List<E> entityList : map.values()) {
-            callables.add(Executors.callable(new Runnable() {
+            tasks.add(Executors.callable(new Runnable() {
                 @Override
                 public void run() {
                     for (final E entity : entityList) {
@@ -781,7 +806,7 @@ public void run() {
                 }
             }));
         }
-        return callables;
+        return tasks;
     }
 
     void recordQueueBinding(AutorecoveringChannel ch,