merge bug26359 into default

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);