Revert the changes to QueueingConsumer since they were considered too ugly and the obnly way to fix would be to replace it with a different API.

Author: Simon MacMullen

src/com/rabbitmq/client/QueueingConsumer.java

@@ -49,15 +49,11 @@ 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, false);
+    private static final Delivery POISON = new Delivery(null, null, null);
 
     public QueueingConsumer(Channel ch) {
         this(ch, new LinkedBlockingQueue<Delivery>());
@@ -81,15 +77,7 @@ public QueueingConsumer(Channel ch, BlockingQueue<Delivery> q) {
         throws IOException
     {
         checkShutdown();
-        this._queue.add(new Delivery(envelope, properties, body,
-                _nextDeliveryRecoverOk));
-        _nextDeliveryRecoverOk = false;
-    }
-
-    @Override
-    public void handleRecoverOk() {
-        checkShutdown();
-        _nextDeliveryRecoverOk = true;
+        this._queue.add(new Delivery(envelope, properties, body));
     }
 
     /**
@@ -99,14 +87,11 @@ 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, boolean recoverOk) {
+        public Delivery(Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
             _envelope = envelope;
             _properties = properties;
             _body = body;
-            _recoverOk = recoverOk;
         }
 
         /**
@@ -132,19 +117,6 @@ 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;
-        }
     }
 
     /**

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

@@ -52,7 +52,6 @@ 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
@@ -65,13 +64,11 @@ 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()));        
@@ -83,7 +80,6 @@ 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);
@@ -96,20 +92,12 @@ 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 {