add RecoveryDelayHandler

Author: vikinghawk

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

@@ -112,6 +112,7 @@ public class ConnectionFactory implements Cloneable {
     // and longs safely. It is unlikely that anybody'd need
     // to use recovery intervals > Integer.MAX_VALUE in practice.
     private long networkRecoveryInterval          = 5000;
+    private RecoveryDelayHandler recoveryDelayHandler;
 
     private MetricsCollector metricsCollector;
 
@@ -960,6 +961,7 @@ public ConnectionParams params(ExecutorService consumerWorkServiceExecutor) {
         result.setShutdownTimeout(shutdownTimeout);
         result.setSaslConfig(saslConfig);
         result.setNetworkRecoveryInterval(networkRecoveryInterval);
+        result.setRecoveryDelayHandler(recoveryDelayHandler);
         result.setTopologyRecovery(topologyRecovery);
         result.setExceptionHandler(exceptionHandler);
         result.setThreadFactory(threadFactory);
@@ -1077,6 +1079,14 @@ public void setNetworkRecoveryInterval(int networkRecoveryInterval) {
     public void setNetworkRecoveryInterval(long networkRecoveryInterval) {
         this.networkRecoveryInterval = networkRecoveryInterval;
     }
+    
+    public RecoveryDelayHandler getRecoveryDelayHandler() {
+        return recoveryDelayHandler;
+    }
+    
+    public void setRecoveryDelayHandler(final RecoveryDelayHandler recoveryDelayHandler) {
+        this.recoveryDelayHandler = recoveryDelayHandler;
+    }
 
     /**
      * Sets the parameters when using NIO.

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

@@ -0,0 +1,46 @@
+package com.rabbitmq.client;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+public interface RecoveryDelayHandler {
+
+    long getDelay(final int recoveryAttempts);
+    
+    public static class DefaultRecoveryDelayHandler implements RecoveryDelayHandler {
+
+        private final long networkRecoveryInterval;
+        
+        public DefaultRecoveryDelayHandler(final long networkRecoveryInterval) {
+            this.networkRecoveryInterval = networkRecoveryInterval;
+        }
+        
+        @Override
+        public long getDelay(int recoveryAttempts) {
+            return networkRecoveryInterval;
+        }
+        
+    }
+    
+    public static class ExponentialBackoffDelayHandler implements RecoveryDelayHandler {
+
+        private final List<Long> sequence;
+        
+        public ExponentialBackoffDelayHandler() {
+            sequence = Arrays.asList(0L, 1000L, 1000L, 2000L, 3000L, 5000L, 8000L, 13000L, 21000L);
+        }
+        
+        public ExponentialBackoffDelayHandler(final List<Long> sequence) {
+            if (sequence.isEmpty())
+                throw new IllegalArgumentException();
+            this.sequence = Collections.unmodifiableList(sequence);
+            
+        }
+        
+        @Override
+        public long getDelay(int recoveryAttempts) {
+            return sequence.get(recoveryAttempts >= sequence.size() ? sequence.size() - 1 : recoveryAttempts);
+        }
+    }
+}

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

@@ -16,6 +16,7 @@
 package com.rabbitmq.client.impl;
 
 import com.rabbitmq.client.ExceptionHandler;
+import com.rabbitmq.client.RecoveryDelayHandler;
 import com.rabbitmq.client.SaslConfig;
 
 import java.util.Map;
@@ -38,6 +39,7 @@ public class ConnectionParams {
     private int shutdownTimeout;
     private SaslConfig saslConfig;
     private long networkRecoveryInterval;
+    private RecoveryDelayHandler recoveryDelayHandler;
     private boolean topologyRecovery;
     private int channelRpcTimeout;
     private boolean channelShouldCheckRpcResponseType;
@@ -102,6 +104,10 @@ public ExceptionHandler getExceptionHandler() {
     public long getNetworkRecoveryInterval() {
         return networkRecoveryInterval;
     }
+    
+    public RecoveryDelayHandler getRecoveryDelayHandler() {
+        return recoveryDelayHandler == null ? new RecoveryDelayHandler.DefaultRecoveryDelayHandler(networkRecoveryInterval) : recoveryDelayHandler;
+    }
 
     public boolean isTopologyRecoveryEnabled() {
         return topologyRecovery;
@@ -162,6 +168,10 @@ public void setSaslConfig(SaslConfig saslConfig) {
     public void setNetworkRecoveryInterval(long networkRecoveryInterval) {
         this.networkRecoveryInterval = networkRecoveryInterval;
     }
+    
+    public void setRecoveryDelayHandler(final RecoveryDelayHandler recoveryDelayHandler) {
+        this.recoveryDelayHandler = recoveryDelayHandler;
+    }
 
     public void setTopologyRecovery(boolean topologyRecovery) {
         this.topologyRecovery = topologyRecovery;

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

@@ -487,7 +487,7 @@ public void removeConsumerRecoveryListener(ConsumerRecoveryListener listener) {
     }
 
     synchronized private void beginAutomaticRecovery() throws InterruptedException {
-        Thread.sleep(this.params.getNetworkRecoveryInterval());
+        Thread.sleep(this.params.getRecoveryDelayHandler().getDelay(0));
 
         this.notifyRecoveryListenersStarted();
 
@@ -525,9 +525,10 @@ private void recoverBlockedListeners(final RecoveryAwareAMQConnection newConn) {
 	// Returns new connection if the connection was recovered, 
 	// null if application initiated shutdown while attempting recovery.  
     private RecoveryAwareAMQConnection recoverConnection() throws InterruptedException {
-        while (!manuallyClosed)
-		{
+        int attempts = 0;
+        while (!manuallyClosed) {
             try {
+                attempts++;
 				RecoveryAwareAMQConnection newConn = this.cf.newConnection();
 				synchronized(recoveryLock) {
 					if (!manuallyClosed) {
@@ -541,8 +542,7 @@ private RecoveryAwareAMQConnection recoverConnection() throws InterruptedExcepti
 				newConn.abort();
 				return null;
             } catch (Exception e) {
-                // TODO: exponential back-off
-                Thread.sleep(this.params.getNetworkRecoveryInterval());
+                Thread.sleep(this.params.getRecoveryDelayHandler().getDelay(attempts));
                 this.getExceptionHandler().handleConnectionRecoveryException(this, e);
             }
         }
@@ -561,13 +561,13 @@ private void recoverChannels(final RecoveryAwareAMQConnection newConn) {
     }
 
     private void notifyRecoveryListenersComplete() {
-        for (RecoveryListener f : this.recoveryListeners) {
+        for (RecoveryListener f : Utility.copy(this.recoveryListeners)) {
             f.handleRecovery(this);
         }
     }
 
     private void notifyRecoveryListenersStarted() {
-        for (RecoveryListener f : this.recoveryListeners) {
+        for (RecoveryListener f : Utility.copy(this.recoveryListeners)) {
             f.handleRecoveryStarted(this);
         }
     }