Polish recovery retry

Add log in default retry handler, add operation to recover all the
bindings of a queue (useful when the recovery of a consumer fails
because isn't found), make AutorecoveringConnection#recoverConsumer and
AutorecoveringConnection#recoverQueue public as they contain useful
logic that some client code should be able to use, and declared a
pre-configured retry handler for the deleted queue case.

References #387

(cherry picked from commit 2b8d257)

Conflicts:
	src/main/java/com/rabbitmq/client/impl/recovery/DefaultRetryHandler.java
	src/main/java/com/rabbitmq/client/impl/recovery/TopologyRecoveryRetryLogic.java

Author: acogoluegnes

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

@@ -718,7 +718,7 @@ public Void call() throws Exception {
     }
 
 
-    void recoverQueue(final String oldName, RecordedQueue q, boolean retry) {
+    public void recoverQueue(final String oldName, RecordedQueue q, boolean retry) {
         try {
             if (topologyRecoveryFilter.filterQueue(q)) {
                 LOGGER.debug("Recovering {}", q);
@@ -789,7 +789,7 @@ public Void call() throws Exception {
         }
     }
 
-    private void recoverConsumer(final String tag, RecordedConsumer consumer, boolean retry) {
+    public void recoverConsumer(final String tag, RecordedConsumer consumer, boolean retry) {
         try {
             if (this.topologyRecoveryFilter.filterConsumer(consumer)) {
                 LOGGER.debug("Recovering {}", consumer);
@@ -1110,6 +1110,10 @@ public Map<String, RecordedExchange> getRecordedExchanges() {
         return recordedExchanges;
     }
 
+    public List<RecordedBinding> getRecordedBindings() {
+        return recordedBindings;
+    }
+
     @Override
     public String toString() {
         return this.delegate.toString();

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

@@ -15,6 +15,9 @@
 
 package com.rabbitmq.client.impl.recovery;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
  * Composable topology recovery retry handler.
  * This retry handler implementations let the user choose the condition
@@ -32,6 +35,8 @@
  */
 public class DefaultRetryHandler implements RetryHandler {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultRetryHandler.class);
+
     private final RetryCondition<? super RecordedQueue> queueRecoveryRetryCondition;
     private final RetryCondition<? super RecordedExchange> exchangeRecoveryRetryCondition;
     private final RetryCondition<? super RecordedBinding> bindingRecoveryRetryCondition;
@@ -91,6 +96,7 @@ public RetryResult retryConsumerRecovery(RetryContext context) throws Exception
 
     protected <T extends RecordedEntity> RetryResult doRetry(RetryCondition<T> condition, RetryOperation<?> operation, T entity, RetryContext context)
         throws Exception {
+        log(entity, context.exception());
         int attempts = 0;
         Exception exception = context.exception();
         while (attempts < retryAttempts) {
@@ -112,6 +118,10 @@ protected <T extends RecordedEntity> RetryResult doRetry(RetryCondition<T> condi
         throw context.exception();
     }
 
+    protected void log(RecordedEntity entity, Exception exception) {
+        LOGGER.info("Error while recovering {}, retrying with {} attempt(s).", entity, retryAttempts, exception);
+    }
+
     public static abstract class RetryOperation<T> {
 
         public abstract T call(RetryContext context) throws Exception;
@@ -157,18 +167,5 @@ public boolean test(E entity, Exception ex) {
                 }
             };
         }
-
-        public RetryCondition<E> or(final RetryCondition<? super E> other) {
-            if (other == null) {
-                throw new IllegalArgumentException("Condition cannot be null");
-            }
-            return new RetryCondition<E>() {
-
-                @Override
-                public boolean test(E entity, Exception ex) {
-                    return RetryCondition.this.test(entity, ex) || other.test(entity, ex);
-                }
-            };
-        }
     }
 }

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

@@ -18,6 +18,8 @@
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.ShutdownSignalException;
 
+import static com.rabbitmq.client.impl.recovery.TopologyRecoveryRetryHandlerBuilder.builder;
+
 /**
  * Useful ready-to-use conditions and operations for {@link DefaultRetryHandler}.
  * They're composed and used with the {@link TopologyRecoveryRetryHandlerBuilder}.
@@ -29,6 +31,9 @@
  */
 public abstract class TopologyRecoveryRetryLogic {
 
+    /**
+     * Channel has been closed because of a resource that doesn't exist.
+     */
     public static final DefaultRetryHandler.RetryCondition<RecordedEntity> CHANNEL_CLOSED_NOT_FOUND = new DefaultRetryHandler.RetryCondition<RecordedEntity>() {
 
         @Override
@@ -43,6 +48,9 @@ public boolean test(RecordedEntity entity, Exception e) {
         }
     };
 
+    /**
+     * Recover a channel.
+     */
     public static final DefaultRetryHandler.RetryOperation<Void> RECOVER_CHANNEL = new DefaultRetryHandler.RetryOperation<Void>() {
 
         @Override
@@ -54,6 +62,9 @@ public Void call(RetryContext context) throws Exception {
         }
     };
 
+    /**
+     * Recover the destination queue of a binding.
+     */
     public static final DefaultRetryHandler.RetryOperation<Void> RECOVER_BINDING_QUEUE = new DefaultRetryHandler.RetryOperation<Void>() {
 
         @Override
@@ -72,6 +83,9 @@ public Void call(RetryContext context) {
         }
     };
 
+    /**
+     * Recover a binding.
+     */
     public static final DefaultRetryHandler.RetryOperation<Void> RECOVER_BINDING = new DefaultRetryHandler.RetryOperation<Void>() {
 
         @Override
@@ -81,6 +95,9 @@ public Void call(RetryContext context) throws Exception {
         }
     };
 
+    /**
+     * Recover the queue of a consumer.
+     */
     public static final DefaultRetryHandler.RetryOperation<Void> RECOVER_CONSUMER_QUEUE = new DefaultRetryHandler.RetryOperation<Void>() {
 
         @Override
@@ -99,11 +116,47 @@ public Void call(RetryContext context) {
         }
     };
 
+    /**
+     * Recover a consumer.
+     */
     public static final DefaultRetryHandler.RetryOperation<String> RECOVER_CONSUMER = new DefaultRetryHandler.RetryOperation<String>() {
 
         @Override
         public String call(RetryContext context) throws Exception {
             return context.consumer().recover();
         }
     };
+
+    /**
+     * Recover all the bindings of the queue of a consumer.
+     */
+    public static final DefaultRetryHandler.RetryOperation<Void> RECOVER_CONSUMER_QUEUE_BINDINGS = new DefaultRetryHandler.RetryOperation<Void>() {
+
+        @Override
+        public Void call(RetryContext context) throws Exception {
+            if (context.entity() instanceof RecordedConsumer) {
+                String queue = context.consumer().getQueue();
+                for (RecordedBinding recordedBinding : context.connection().getRecordedBindings()) {
+                    if (recordedBinding instanceof RecordedQueueBinding && queue.equals(recordedBinding.getDestination())) {
+                        recordedBinding.recover();
+                    }
+                }
+            }
+            return null;
+        }
+    };
+
+    /**
+     * Pre-configured {@link DefaultRetryHandler} that retries recovery of bindings and consumers
+     * when their respective queue is not found.
+     * This retry handler can be useful for long recovery processes, whereby auto-delete queues
+     * can be deleted between queue recovery and binding/consumer recovery.
+     */
+    public static final RetryHandler RETRY_ON_QUEUE_NOT_FOUND_RETRY_HANDLER = builder()
+        .bindingRecoveryRetryCondition(CHANNEL_CLOSED_NOT_FOUND)
+        .consumerRecoveryRetryCondition(CHANNEL_CLOSED_NOT_FOUND)
+        .bindingRecoveryRetryOperation(RECOVER_CHANNEL.andThen(RECOVER_BINDING_QUEUE).andThen(RECOVER_BINDING))
+        .consumerRecoveryRetryOperation(RECOVER_CHANNEL.andThen(RECOVER_CONSUMER_QUEUE.andThen(RECOVER_CONSUMER)
+            .andThen(RECOVER_CONSUMER_QUEUE_BINDINGS)))
+        .build();
 }

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

@@ -27,13 +27,7 @@
 
 import java.util.HashMap;
 
-import static com.rabbitmq.client.impl.recovery.TopologyRecoveryRetryHandlerBuilder.builder;
-import static com.rabbitmq.client.impl.recovery.TopologyRecoveryRetryLogic.CHANNEL_CLOSED_NOT_FOUND;
-import static com.rabbitmq.client.impl.recovery.TopologyRecoveryRetryLogic.RECOVER_BINDING;
-import static com.rabbitmq.client.impl.recovery.TopologyRecoveryRetryLogic.RECOVER_BINDING_QUEUE;
-import static com.rabbitmq.client.impl.recovery.TopologyRecoveryRetryLogic.RECOVER_CHANNEL;
-import static com.rabbitmq.client.impl.recovery.TopologyRecoveryRetryLogic.RECOVER_CONSUMER;
-import static com.rabbitmq.client.impl.recovery.TopologyRecoveryRetryLogic.RECOVER_CONSUMER_QUEUE;
+import static com.rabbitmq.client.impl.recovery.TopologyRecoveryRetryLogic.RETRY_ON_QUEUE_NOT_FOUND_RETRY_HANDLER;
 import static com.rabbitmq.client.test.TestUtils.closeAllConnectionsAndWaitForRecovery;
 import static org.junit.Assert.assertTrue;
 
@@ -61,13 +55,7 @@ public void topologyRecoveryRetry() throws Exception {
     @Override
     protected ConnectionFactory newConnectionFactory() {
         ConnectionFactory connectionFactory = TestUtils.connectionFactory();
-        connectionFactory.setTopologyRecoveryRetryHandler(
-            builder().bindingRecoveryRetryCondition(CHANNEL_CLOSED_NOT_FOUND)
-                .consumerRecoveryRetryCondition(CHANNEL_CLOSED_NOT_FOUND)
-                .bindingRecoveryRetryOperation(RECOVER_CHANNEL.andThen(RECOVER_BINDING_QUEUE).andThen(RECOVER_BINDING))
-                .consumerRecoveryRetryOperation(RECOVER_CHANNEL.andThen(RECOVER_CONSUMER_QUEUE.andThen(RECOVER_CONSUMER)))
-                .build()
-        );
+        connectionFactory.setTopologyRecoveryRetryHandler(RETRY_ON_QUEUE_NOT_FOUND_RETRY_HANDLER);
         connectionFactory.setNetworkRecoveryInterval(1000);
         return connectionFactory;
     }