Merge pull request #370 from vikinghawk/4.x.x-stable

Multi-threaded topology recovery

Author: acogoluegnes

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

@@ -129,7 +129,8 @@ public class ConnectionFactory implements Cloneable {
 
     private boolean automaticRecovery               = true;
     private boolean topologyRecovery                = true;
-
+    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
     // to use recovery intervals > Integer.MAX_VALUE in practice.
@@ -341,7 +342,7 @@ public void setUri(String uriString)
         setUri(new URI(uriString));
     }
 
-    private String uriDecode(String s) {
+    private static String uriDecode(String s) {
         try {
             // URLDecode decodes '+' to a space, as for
             // form encoding.  So protect plus signs.
@@ -525,7 +526,6 @@ public void setSocketFactory(SocketFactory factory) {
      *
      * @see #setSocketConfigurator(SocketConfigurator)
      */
-    @SuppressWarnings("unused")
     public SocketConfigurator getSocketConfigurator() {
         return socketConf;
     }
@@ -703,7 +703,6 @@ public void setAutomaticRecoveryEnabled(boolean automaticRecovery) {
      * @return true if topology recovery is enabled, false otherwise
      * @see <a href="http://www.rabbitmq.com/api-guide.html#recovery">Automatic Recovery</a>
      */
-    @SuppressWarnings("unused")
     public boolean isTopologyRecoveryEnabled() {
         return topologyRecovery;
     }
@@ -716,6 +715,24 @@ public boolean isTopologyRecoveryEnabled() {
     public void setTopologyRecoveryEnabled(boolean topologyRecovery) {
         this.topologyRecovery = topologyRecovery;
     }
+    
+    /**
+     * 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;
+    }
+    
+    /**
+     * 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) {
         this.metricsCollector = metricsCollector;
@@ -1015,6 +1032,7 @@ public ConnectionParams params(ExecutorService consumerWorkServiceExecutor) {
         result.setNetworkRecoveryInterval(networkRecoveryInterval);
         result.setRecoveryDelayHandler(recoveryDelayHandler);
         result.setTopologyRecovery(topologyRecovery);
+        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,6 +41,7 @@ public class ConnectionParams {
     private long networkRecoveryInterval;
     private RecoveryDelayHandler recoveryDelayHandler;
     private boolean topologyRecovery;
+    private ExecutorService topologyRecoveryExecutor;
     private int channelRpcTimeout;
     private boolean channelShouldCheckRpcResponseType;
     private ErrorOnWriteListener errorOnWriteListener;
@@ -114,10 +115,18 @@ public RecoveryDelayHandler getRecoveryDelayHandler() {
     public boolean isTopologyRecoveryEnabled() {
         return topologyRecovery;
     }
+    
+    /**
+     * 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() {
-    return threadFactory;
-  }
+        return threadFactory;
+    }
 
     public int getChannelRpcTimeout() {
         return channelRpcTimeout;
@@ -174,6 +183,10 @@ public void setRecoveryDelayHandler(final RecoveryDelayHandler recoveryDelayHand
     public void setTopologyRecovery(boolean topologyRecovery) {
         this.topologyRecovery = topologyRecovery;
     }
+    
+    public void setTopologyRecoveryExecutor(final ExecutorService topologyRecoveryExecutor) {
+        this.topologyRecoveryExecutor = topologyRecoveryExecutor;
+    }
 
     public void setExceptionHandler(ExceptionHandler exceptionHandler) {
         this.exceptionHandler = exceptionHandler;