Introduce listeners for queue name and consumer tag changes (during recovery)

Meant to be used by clients like March Hare that represent queues and consumers
as object and may need to update them, while relying on the Java client
to perform recovery.

Author: michaelklishin

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

@@ -60,6 +60,8 @@ public class AutorecoveringConnection implements Connection, Recoverable, Networ
     private final List<RecordedBinding> recordedBindings = new ArrayList<RecordedBinding>();
     private Map<String, RecordedExchange> recordedExchanges = new ConcurrentHashMap<String, RecordedExchange>();
     private final Map<String, RecordedConsumer> consumers = new ConcurrentHashMap<String, RecordedConsumer>();
+    private List<ConsumerRecoveryListener> consumerRecoveryListeners = new ArrayList<ConsumerRecoveryListener>();
+    private List<QueueRecoveryListener> queueRecoveryListeners = new ArrayList<QueueRecoveryListener>();
 
     public AutorecoveringConnection(ConnectionParams params, FrameHandlerFactory f, Address[] addrs) {
         this.cf = new RecoveryAwareAMQConnectionFactory(params, f, addrs);
@@ -355,6 +357,44 @@ public void shutdownCompleted(ShutdownSignalException cause) {
         }
     }
 
+    /**
+     * Not part of the public API. Mean to be used by JVM RabbitMQ clients that build on
+     * top of the Java client and need to be notified when server-named queue name changes
+     * after recovery.
+     *
+     * @param listener listener that observes queue name changes after recovery
+     */
+    public void addQueueRecoveryListener(QueueRecoveryListener listener) {
+        this.queueRecoveryListeners.add(listener);
+    }
+
+    /**
+     * @see com.rabbitmq.client.impl.recovery.AutorecoveringConnection#addQueueRecoveryListener
+     * @param listener listener to be removed
+     */
+    public void removeQueueRecoveryListener(QueueRecoveryListener listener) {
+        this.queueRecoveryListeners.remove(listener);
+    }
+
+    /**
+     * Not part of the public API. Mean to be used by JVM RabbitMQ clients that build on
+     * top of the Java client and need to be notified when consumer tag changes
+     * after recovery.
+     *
+     * @param listener listener that observes consumer tag changes after recovery
+     */
+    public void addConsumerRecoveryListener(ConsumerRecoveryListener listener) {
+        this.consumerRecoveryListeners.add(listener);
+    }
+
+    /**
+     * @see com.rabbitmq.client.impl.recovery.AutorecoveringConnection#addConsumerRecoveryListener(ConsumerRecoveryListener)
+     * @param listener listener to be removed
+     */
+    public void removeConsumerRecoveryListener(ConsumerRecoveryListener listener) {
+        this.consumerRecoveryListeners.remove(listener);
+    }
+
     synchronized private void beginAutomaticRecovery() throws InterruptedException, IOException, TopologyRecoveryException {
         Thread.sleep(this.params.getNetworkRecoveryInterval());
         this.recoverConnection();
@@ -455,6 +495,9 @@ private void recoverQueues() {
                     this.propagateQueueNameChangeToBindings(oldName, newName);
                     this.propagateQueueNameChangeToConsumers(oldName, newName);
                 }
+                for(QueueRecoveryListener qrl : this.queueRecoveryListeners) {
+                    qrl.queueRecovered(oldName, newName);
+                }
             } catch (Exception cause) {
                 final String message = "Caught an exception while recovering queue " + oldName +
                                                ": " + cause.getMessage();
@@ -490,6 +533,9 @@ private void recoverConsumers() {
                     this.consumers.remove(tag);
                     this.consumers.put(newTag, consumer);
                 }
+                for(ConsumerRecoveryListener crl : this.consumerRecoveryListeners) {
+                    crl.consumerRecovered(tag, newTag);
+                }
             } catch (Exception cause) {
                 final String message = "Caught an exception while recovering consumer " + tag +
                         ": " + cause.getMessage();

src/com/rabbitmq/client/impl/recovery/ConsumerRecoveryListener.java

@@ -0,0 +1,10 @@
+package com.rabbitmq.client.impl.recovery;
+
+/**
+ * Not part of the public API. Mean to be used by JVM RabbitMQ clients that build on
+ * top of the Java client and need to be notified when consumer tag changes
+ * after recovery.
+ */
+public interface ConsumerRecoveryListener {
+    void consumerRecovered(String oldConsumerTag, String newConsumerTag);
+}

src/com/rabbitmq/client/impl/recovery/QueueRecoveryListener.java

@@ -0,0 +1,10 @@
+package com.rabbitmq.client.impl.recovery;
+
+/**
+ * Not part of the public API. Mean to be used by JVM RabbitMQ clients that build on
+ * top of the Java client and need to be notified when server-name queue name changes
+ * after recovery.
+ */
+public interface QueueRecoveryListener {
+    void queueRecovered(String oldName, String newName);
+}

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

@@ -5,6 +5,8 @@
 import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
 import com.rabbitmq.client.Recoverable;
 import com.rabbitmq.client.RecoveryListener;
+import com.rabbitmq.client.impl.recovery.ConsumerRecoveryListener;
+import com.rabbitmq.client.impl.recovery.QueueRecoveryListener;
 import com.rabbitmq.client.test.BrokerTestCase;
 import com.rabbitmq.tools.Host;
 
@@ -16,6 +18,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 public class ConnectionRecovery extends BrokerTestCase {
     public static final long RECOVERY_INTERVAL = 2000;
@@ -208,10 +211,24 @@ public void testServerNamedQueueRecovery() throws IOException, InterruptedExcept
         String x = "amq.fanout";
         channel.queueBind(q, x, "");
 
+        final AtomicReference<String> nameBefore = new AtomicReference<String>();
+        final AtomicReference<String> nameAfter  = new AtomicReference<String>();
+        final CountDownLatch listenerLatch = new CountDownLatch(1);
+        ((AutorecoveringConnection)connection).addQueueRecoveryListener(new QueueRecoveryListener() {
+            @Override
+            public void queueRecovered(String oldName, String newName) {
+                nameBefore.set(oldName);
+                nameAfter.set(newName);
+                listenerLatch.countDown();
+            }
+        });
+
         closeAndWaitForRecovery();
+        wait(listenerLatch);
         expectChannelRecovery(channel);
         channel.basicPublish(x, "", null, "msg".getBytes());
         assertDelivered(q, 1);
+        assertFalse(nameBefore.get().equals(nameAfter.get()));
         channel.queueDelete(q);
     }
 
@@ -318,8 +335,22 @@ public void testConsumerRecoveryWithManyConsumers() throws IOException, Interrup
         for (int i = 0; i < n; i++) {
             channel.basicConsume(q, new DefaultConsumer(channel));
         }
+        final AtomicReference<String> tagA = new AtomicReference<String>();
+        final AtomicReference<String> tagB = new AtomicReference<String>();
+        final CountDownLatch listenerLatch = new CountDownLatch(n);
+        ((AutorecoveringConnection)connection).addConsumerRecoveryListener(new ConsumerRecoveryListener() {
+            @Override
+            public void consumerRecovered(String oldConsumerTag, String newConsumerTag) {
+                tagA.set(oldConsumerTag);
+                tagB.set(newConsumerTag);
+                listenerLatch.countDown();
+            }
+        });
+
         assertConsumerCount(n, q);
         closeAndWaitForRecovery();
+        wait(listenerLatch);
+        assertTrue(tagA.get().equals(tagB.get()));
         expectChannelRecovery(channel);
         assertConsumerCount(n, q);