Merging bug 22421 into amqp_0_9_1

Author: Matthew Sackman

src/com/rabbitmq/client/Consumer.java

@@ -70,6 +70,11 @@ public interface Consumer {
      */
     void handleShutdownSignal(String consumerTag, ShutdownSignalException sig);
 
+    /**
+     *
+     */
+    void handleRecoverOk();
+
     /**
      * Called when a delivery appears for this consumer.
      * @param consumerTag the defined consumerTag (either client- or server-generated)

src/com/rabbitmq/client/DefaultConsumer.java

@@ -74,6 +74,13 @@ public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig
         // no work to do
     }
 
+    /**
+     * No-op implementation of {@link Consumer#handleRecoverOk}.
+     */
+    public void handleRecoverOk() {
+        // no work to do
+    }
+
     /**
      * No-op implementation of {@link Consumer#handleDelivery}.
      */

src/com/rabbitmq/client/impl/ChannelN.java

@@ -258,6 +258,15 @@ public void releaseChannelNumber() {
                     _channelMutex.notifyAll();
                 }
                 return true;
+            } else if (method instanceof Basic.RecoverOk) {
+                for (Consumer callback: _consumers.values()) {
+                    callback.handleRecoverOk();
+                }
+
+                // Unlike all the other cases we still want this RecoverOk to
+                // be handled by whichever RPC continuation invoked Recover,
+                // so return false
+                return false;
             } else {
                 return false;
             }