Recover channel shutdown hooks

Author: michaelklishin

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

@@ -30,6 +30,7 @@
 public class AutorecoveringChannel implements Channel, Recoverable {
     private RecoveryAwareChannelN delegate;
     private AutorecoveringConnection connection;
+    private final List<ShutdownListener> shutdownHooks  = new ArrayList<ShutdownListener>();
     private List<RecoveryListener> recoveryListeners = new ArrayList<RecoveryListener>();
     private List<ReturnListener> returnListeners = new ArrayList<ReturnListener>();
     private List<ConfirmListener> confirmListeners = new ArrayList<ConfirmListener>();
@@ -379,11 +380,16 @@ public Command rpc(Method method) throws IOException {
         return delegate.rpc(method);
     }
 
+    /**
+     * @see Connection#addShutdownListener(com.rabbitmq.client.ShutdownListener)
+     */
     public void addShutdownListener(ShutdownListener listener) {
+        this.shutdownHooks.add(listener);
         delegate.addShutdownListener(listener);
     }
 
     public void removeShutdownListener(ShutdownListener listener) {
+        this.shutdownHooks.remove(listener);
         delegate.removeShutdownListener(listener);
     }
 
@@ -417,13 +423,20 @@ public void automaticallyRecover(AutorecoveringConnection connection, Connection
         this.delegate = (RecoveryAwareChannelN) connDelegate.createChannel(this.getChannelNumber());
         this.delegate.inheritOffsetFrom(defunctChannel);
 
+        this.recoverShutdownListeners();
         this.recoverReturnListeners();
         this.recoverConfirmListeners();
         this.recoverFlowListeners();
         this.recoverState();
         this.notifyRecoveryListeners();
     }
 
+    private void recoverShutdownListeners() {
+        for (ShutdownListener sh : this.shutdownHooks) {
+            this.delegate.addShutdownListener(sh);
+        }
+    }
+
     private void recoverReturnListeners() {
         for(ReturnListener rl : this.returnListeners) {
             this.delegate.addReturnListener(rl);

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

@@ -68,7 +68,7 @@ public void testConnectionRecoveryWithDisabledTopologyRecovery() throws IOExcept
         }
     }
 
-    public void testShutdownHooksRecovery() throws IOException, InterruptedException {
+    public void testShutdownHooksRecoveryOnConnection() throws IOException, InterruptedException {
         final CountDownLatch latch = new CountDownLatch(2);
         connection.addShutdownListener(new ShutdownListener() {
             public void shutdownCompleted(ShutdownSignalException cause) {
@@ -82,6 +82,22 @@ public void shutdownCompleted(ShutdownSignalException cause) {
         wait(latch);
     }
 
+    public void testShutdownHooksRecoveryOnChannel() throws IOException, InterruptedException {
+        final CountDownLatch latch = new CountDownLatch(3);
+        channel.addShutdownListener(new ShutdownListener() {
+            public void shutdownCompleted(ShutdownSignalException cause) {
+                latch.countDown();
+            }
+        });
+        assertTrue(connection.isOpen());
+        closeAndWaitForRecovery();
+        assertTrue(connection.isOpen());
+        closeAndWaitForRecovery();
+        assertTrue(connection.isOpen());
+        connection.close();
+        wait(latch);
+    }
+
     public void testBlockedListenerRecovery() throws IOException, InterruptedException {
         final CountDownLatch latch = new CountDownLatch(2);
         connection.addBlockedListener(new BlockedListener() {