Revert "recover channels in parallel too"

This reverts commit 774590a.

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 recoveryThreads                     = 1;
+    private int topologyRecoveryThreads             = 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 getRecoveryThreadCount() {
-        return recoveryThreads;
+    public int getTopologyRecoveryThreadCount() {
+        return topologyRecoveryThreads;
     }
 
     // TODO Document that your exception handler method should be thread safe
-    public void setRecoveryThreadCount(final int recoveryThreads) {
-        this.recoveryThreads = recoveryThreads;
+    public void setTopologyRecoveryThreadCount(final int topologyRecoveryThreads) {
+        this.topologyRecoveryThreads = topologyRecoveryThreads;
     }
 
     public void setMetricsCollector(MetricsCollector metricsCollector) {
@@ -1021,7 +1021,7 @@ public ConnectionParams params(ExecutorService consumerWorkServiceExecutor) {
         result.setNetworkRecoveryInterval(networkRecoveryInterval);
         result.setRecoveryDelayHandler(recoveryDelayHandler);
         result.setTopologyRecovery(topologyRecovery);
-        result.setRecoveryThreadCount(recoveryThreads);
+        result.setTopologyRecoveryThreadCount(topologyRecoveryThreads);
         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 recoveryThreads = 1;
+    private int topologyRecoveryThreads = 1;
     private int channelRpcTimeout;
     private boolean channelShouldCheckRpcResponseType;
     private ErrorOnWriteListener errorOnWriteListener;
@@ -116,8 +116,8 @@ public boolean isTopologyRecoveryEnabled() {
         return topologyRecovery;
     }
 
-    public int getRecoveryThreadCount() {
-        return recoveryThreads;
+    public int getTopologyRecoveryThreadCount() {
+        return topologyRecoveryThreads;
     }
 
     public ThreadFactory getThreadFactory() {
@@ -180,8 +180,8 @@ public void setTopologyRecovery(boolean topologyRecovery) {
         this.topologyRecovery = topologyRecovery;
     }
 
-    public void setRecoveryThreadCount(final int recoveryThreads) {
-        this.recoveryThreads = recoveryThreads;
+    public void setTopologyRecoveryThreadCount(final int topologyRecoveryThreads) {
+        this.topologyRecoveryThreads = topologyRecoveryThreads;
     }
 
     public void setExceptionHandler(ExceptionHandler exceptionHandler) {

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

@@ -540,25 +540,12 @@ private synchronized void beginAutomaticRecovery() throws InterruptedException {
         this.addAutomaticRecoveryListener(newConn);
 	    this.recoverShutdownListeners(newConn);
 	    this.recoverBlockedListeners(newConn);
-	    
-	    // 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.recoverChannels(newConn);
+	    // don't assign new delegate connection until channel recovery is complete
+	    this.delegate = newConn;
+	    if (this.params.isTopologyRecoveryEnabled()) {
+	        recoverTopology(params.getTopologyRecoveryThreadCount());
+	    }
 		this.notifyRecoveryListenersComplete();
     }
 
@@ -602,34 +589,16 @@ private RecoveryAwareAMQConnection recoverConnection() throws InterruptedExcepti
 		return null;
     }
 
-    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 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 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)) {
@@ -643,20 +612,26 @@ private void notifyRecoveryListenersStarted() {
         }
     }
 
-    private void recoverTopology(final ExecutorService executor) throws InterruptedException {
+    private void recoverTopology(final int recoveryThreads) throws InterruptedException {
         // The recovery sequence is the following:
         // 1. Recover exchanges
         // 2. Recover queues
         // 3. Recover bindings
         // 4. Recover consumers
-        if (executor != null) {
+        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
-            // 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()));
+            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 {
             // recover entities in serial on the main connection thread
             for (final RecordedExchange exchange : Utility.copy(recordedExchanges).values()) {
@@ -785,9 +760,9 @@ private <E extends RecordedEntity> List<Callable<Object>> groupEntitiesByChannel
             list.add(entity);
         }
         // now create a runnable per channel
-        final List<Callable<Object>> tasks = new ArrayList<Callable<Object>>();
+        final List<Callable<Object>> callables = new ArrayList<Callable<Object>>();
         for (final List<E> entityList : map.values()) {
-            tasks.add(Executors.callable(new Runnable() {
+            callables.add(Executors.callable(new Runnable() {
                 @Override
                 public void run() {
                     for (final E entity : entityList) {
@@ -806,7 +781,7 @@ public void run() {
                 }
             }));
         }
-        return tasks;
+        return callables;
     }
 
     void recordQueueBinding(AutorecoveringChannel ch,