Make available the knowledge that we've just seen a basic.recover_ok to the thread that's handling Consumer callbacks, and in QueueingConsumer.

Author: simonmacmullen

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/QueueingConsumer.java

@@ -49,11 +49,15 @@ public class QueueingConsumer extends DefaultConsumer {
     // throw a shutdown signal exception.
     private volatile ShutdownSignalException _shutdown;
 
+    // When this is true we have seen recover_ok but not had a delivery to
+    // attach it to
+    private boolean _nextDeliveryRecoverOk = false;
+
     // Marker object used to signal the queue is in shutdown mode.
     // It is only there to wake up consumers. The canonical representation
     // of shutting down is the presence of _shutdown.
     // Invariant: This is never on _queue unless _shutdown != null.
-    private static final Delivery POISON = new Delivery(null, null, null);
+    private static final Delivery POISON = new Delivery(null, null, null, false);
 
     public QueueingConsumer(Channel ch) {
         this(ch, new LinkedBlockingQueue<Delivery>());
@@ -77,7 +81,15 @@ public QueueingConsumer(Channel ch, BlockingQueue<Delivery> q) {
         throws IOException
     {
         checkShutdown();
-        this._queue.add(new Delivery(envelope, properties, body));
+        this._queue.add(new Delivery(envelope, properties, body,
+                _nextDeliveryRecoverOk));
+        _nextDeliveryRecoverOk = false;
+    }
+
+    @Override
+    public void handleRecoverOk() {
+        checkShutdown();
+        _nextDeliveryRecoverOk = true;
     }
 
     /**
@@ -87,11 +99,14 @@ public static class Delivery {
         private final Envelope _envelope;
         private final AMQP.BasicProperties _properties;
         private final byte[] _body;
+        private final boolean _recoverOk;
 
-        public Delivery(Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
+        public Delivery(Envelope envelope, AMQP.BasicProperties properties,
+                        byte[] body, boolean recoverOk) {
             _envelope = envelope;
             _properties = properties;
             _body = body;
+            _recoverOk = recoverOk;
         }
 
         /**
@@ -117,6 +132,19 @@ public BasicProperties getProperties() {
         public byte[] getBody() {
             return _body;
         }
+
+        /**
+         * Have we seen recover_ok immediately before this message?
+         * If true, then all messages <b>before</b> this one that have not been
+         * acked will be recovered, but this one and subsequent ones will not.
+         *
+         * NB: this method does <b>not</b> tell you whether this message, or
+         * any other, actually is recovered - for that, see Envelope.isRedeliver().
+         * @return whether we have seen recover_ok immediately before this message
+         */
+        public boolean getRecoverOk() {
+            return _recoverOk;
+        }
     }
 
     /**

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

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

@@ -52,6 +52,7 @@ public void createResources() throws IOException {
 
     static interface RecoverCallback {
         void recover(Channel channel) throws IOException;
+        void checkRecover(QueueingConsumer.Delivery delivery, boolean expected) throws IOException;
     }
 
     // The AMQP specification under-specifies the behaviour when
@@ -64,11 +65,13 @@ void verifyRedeliverOnRecover(RecoverCallback call)
         channel.basicConsume(queue, false, consumer); // require acks.
         channel.basicPublish("", queue, new AMQP.BasicProperties(), body);
         QueueingConsumer.Delivery delivery = consumer.nextDelivery();
+        call.checkRecover(delivery, false);
         assertTrue("consumed message body not as sent",
                    Arrays.equals(body, delivery.getBody()));
         // Don't ack it, and get it redelivered to the same consumer
         call.recover(channel);
         QueueingConsumer.Delivery secondDelivery = consumer.nextDelivery(5000);
+        call.checkRecover(secondDelivery, true);
         assertNotNull("timed out waiting for redelivered message", secondDelivery);
         assertTrue("consumed (redelivered) message body not as sent",
                    Arrays.equals(body, delivery.getBody()));        
@@ -80,6 +83,7 @@ void verifyNoRedeliveryWithAutoAck(RecoverCallback call)
         channel.basicConsume(queue, true, consumer); // auto ack.
         channel.basicPublish("", queue, new AMQP.BasicProperties(), body);
         QueueingConsumer.Delivery delivery = consumer.nextDelivery();
+        call.checkRecover(delivery, false);
         assertTrue("consumed message body not as sent",
                    Arrays.equals(body, delivery.getBody()));
         call.recover(channel);
@@ -92,12 +96,20 @@ void verifyNoRedeliveryWithAutoAck(RecoverCallback call)
             public void recover(Channel channel) throws IOException {
                 channel.basicRecoverAsync(true);
             }
+
+            public void checkRecover(QueueingConsumer.Delivery delivery, boolean expected) throws IOException {
+                // We make no guarantees
+            }
         };
 
     RecoverCallback recoverSync = new RecoverCallback() {
             public void recover(Channel channel) throws IOException {
                 channel.basicRecover(true);
             }
+
+            public void checkRecover(QueueingConsumer.Delivery delivery, boolean expected) throws IOException {
+                assertEquals(expected, delivery.getRecoverOk());
+            }
         };
 
     public void testRedeliverOnRecoverAsync() throws IOException, InterruptedException {