Merge bug21844 into default

Author: tonyg

src/com/rabbitmq/client/Channel.java

@@ -468,6 +468,18 @@ Queue.DeclareOk queueDeclare(String queue, boolean passive, boolean durable, boo
      */
     void basicCancel(String consumerTag) throws IOException;
 
+    /**
+     * Ask the broker to resend unacknowledged messages.  In 0-8
+     * basic.recover is asynchronous; in 0-9-1 it is synchronous, and
+     * the new, deprecated method basic.recover_async is asynchronous.
+     * To avoid this API changing, this is named for the latter, and
+     * will be deprecated.
+     * @param requeue If true, messages will be requeued and possibly
+     * delivered to a different consumer. If false, messages will be
+     * redelivered to the same consumer.
+     */
+    void basicRecoverAsync(boolean requeue) throws IOException;
+
     /**
      * Enables TX mode on this channel.
      * @see com.rabbitmq.client.AMQP.Tx.Select

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

@@ -668,6 +668,13 @@ public Consumer transformReply(AMQCommand replyCommand) {
         }
     }
 
+    /** Public API - {@inheritDoc} */
+    public void basicRecoverAsync(boolean requeue)
+        throws IOException
+    {
+        transmit(new Basic.Recover(requeue));
+    }
+  
     /** Public API - {@inheritDoc} */
     public Tx.SelectOk txSelect()
         throws IOException

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

@@ -41,6 +41,7 @@ public static TestSuite suite() {
         suite.addTestSuite(DoubleDeletion.class);
         suite.addTestSuite(Routing.class);
         suite.addTestSuite(BindingLifecycle.class);
+        suite.addTestSuite(Recover.class);
         suite.addTestSuite(Transactions.class);
         suite.addTestSuite(PersistentTransactions.class);
         suite.addTestSuite(RequeueOnConnectionClose.class);

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

@@ -0,0 +1,80 @@
+//   The contents of this file are subject to the Mozilla Public License
+//   Version 1.1 (the "License"); you may not use this file except in
+//   compliance with the License. You may obtain a copy of the License at
+//   http://www.mozilla.org/MPL/
+//
+//   Software distributed under the License is distributed on an "AS IS"
+//   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+//   License for the specific language governing rights and limitations
+//   under the License.
+//
+//   The Original Code is RabbitMQ.
+//
+//   The Initial Developers of the Original Code are LShift Ltd,
+//   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+//   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+//   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+//   Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift
+//   Ltd. Portions created by Cohesive Financial Technologies LLC are
+//   Copyright (C) 2007-2009 Cohesive Financial Technologies
+//   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+//   (C) 2007-2009 Rabbit Technologies Ltd.
+//
+//   All Rights Reserved.
+//
+//   Contributor(s): ______________________________________.
+//
+
+package com.rabbitmq.client.test.functional;
+
+import java.util.Arrays;
+import java.io.IOException;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.QueueingConsumer;
+
+public class Recover extends BrokerTestCase {
+
+    String queue;
+    byte[] body = "message".getBytes();
+    
+    public void createResources() throws IOException {
+        AMQP.Queue.DeclareOk ok = channel.queueDeclare();
+        queue = ok.getQueue();
+    }
+
+    public void testRedeliverOnRecover() throws IOException, InterruptedException {
+        QueueingConsumer consumer = new QueueingConsumer(channel);
+        channel.basicConsume(queue, false, consumer); // require acks.
+        channel.basicPublish("", queue, new AMQP.BasicProperties(), body);
+        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
+        assertTrue("consumed message body not as sent",
+                   Arrays.equals(body, delivery.getBody()));
+        // Don't ack it, and get it redelivered to the same consumer
+        channel.basicRecoverAsync(false);
+        QueueingConsumer.Delivery secondDelivery = consumer.nextDelivery(5000);
+        assertNotNull("timed out waiting for redelivered message", secondDelivery);
+        assertTrue("consumed (redelivered) message body not as sent",
+                   Arrays.equals(body, delivery.getBody()));        
+    }
+
+    public void testNoRedeliveryWithAutoAck() throws IOException, InterruptedException {
+        QueueingConsumer consumer = new QueueingConsumer(channel);
+        channel.basicConsume(queue, true, consumer); // auto ack.
+        channel.basicPublish("", queue, new AMQP.BasicProperties(), body);
+        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
+        assertTrue("consumed message body not as sent",
+                   Arrays.equals(body, delivery.getBody()));
+        channel.basicRecoverAsync(false);
+        // there's a race here between our recover finishing and the basic.get;
+        Thread.sleep(500);
+        assertNull("should be no message available", channel.basicGet(queue, true));
+    }
+
+    // The AMQP specification under-specifies the behaviour when
+    // requeue=false.  So we can't really test anything.
+}