Merged bug22864 into default

Author: Emile Joubert

build.xml

@@ -26,7 +26,7 @@
     <pathelement path="${test.javac.out}"/>
   </path>
 
-  <property name="AMQP_SPEC_JSON_PATH" value="${codegen.dir}/amqp-${spec.version}.json"/>
+  <property name="AMQP_SPEC_JSON_PATH" value="${codegen.dir}/amqp-${spec.version}.json ${codegen.dir}/rabbitmq-0.8-extensions.json"/>
 
   <target name="amqp-generate-check" description="check if codegen needs to be run">
     <uptodate property="amqp.generate.notRequired">

src/com/rabbitmq/client/Channel.java

@@ -495,12 +495,22 @@ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, b
      * 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.
      */
+    Basic.RecoverOk basicRecover(boolean requeue) 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
+     * and 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.
+     */
+    @Deprecated
     void basicRecoverAsync(boolean requeue) throws IOException;
 
     /**

src/com/rabbitmq/client/Consumer.java

@@ -70,6 +70,19 @@ public interface Consumer {
      */
     void handleShutdownSignal(String consumerTag, ShutdownSignalException sig);
 
+    /**
+     * Called to notify the consumer that we've received a basic.recover-ok
+     * in reply to a basic.recover some other thread sent. All messages
+     * received before this is invoked that haven't been ack'ed will be
+     * redelivered. All messages received afterwards won't be.
+     *
+     * This method exists since all the Consumer callbacks are invoked by the
+     * connection main loop thread - so it's sometimes useful to allow that
+     * thread to know that the recover-ok has been received, rather than the
+     * thread which invoked basicRecover().
+     */
+    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

@@ -285,6 +285,15 @@ public void releaseChannelNumber() {
                     }
                 }
                 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;
             }
@@ -699,11 +708,19 @@ public Consumer transformReply(AMQCommand replyCommand) {
         }
     }
 
+     /** Public API - {@inheritDoc} */
+    public Basic.RecoverOk basicRecover(boolean requeue)
+        throws IOException
+    {
+        return (Basic.RecoverOk) exnWrappingRpc(new Basic.Recover(requeue)).getMethod();
+    }
+
+
     /** Public API - {@inheritDoc} */
     public void basicRecoverAsync(boolean requeue)
         throws IOException
     {
-        transmit(new Basic.Recover(requeue));
+        transmit(new Basic.RecoverAsync(requeue));
     }
 
     /** Public API - {@inheritDoc} */

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

@@ -36,48 +36,85 @@
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.QueueingConsumer;
+import com.rabbitmq.client.Channel;
 
 import com.rabbitmq.client.test.BrokerTestCase;
 
 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 {
+    static interface RecoverCallback {
+        void recover(Channel channel) throws IOException;
+    }
+
+    // The AMQP specification under-specifies the behaviour when
+    // requeue=false.  So we can't really test any scenarios for
+    // requeue=false.
+
+    void verifyRedeliverOnRecover(RecoverCallback call)
+        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(true);
+        call.recover(channel);
         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()));        
+                   Arrays.equals(body, delivery.getBody()));
     }
 
-    public void testNoRedeliveryWithAutoAck() throws IOException, InterruptedException {
+    void verifyNoRedeliveryWithAutoAck(RecoverCallback call)
+        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(true);
+        call.recover(channel);
         // 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 any scenarios for
-    // requeue=false.
+    RecoverCallback recoverAsync = new RecoverCallback() {
+            public void recover(Channel channel) throws IOException {
+                channel.basicRecoverAsync(true);
+            }
+        };
+
+    RecoverCallback recoverSync = new RecoverCallback() {
+            public void recover(Channel channel) throws IOException {
+                channel.basicRecover(true);
+            }
+        };
+
+    public void testRedeliverOnRecoverAsync() throws IOException, InterruptedException {
+        verifyRedeliverOnRecover(recoverAsync);
+    }
+
+    public void testRedeliveryOnRecover() throws IOException, InterruptedException {
+        verifyRedeliverOnRecover(recoverSync);
+    }
+
+    public void testNoRedeliveryWithAutoAckAsync()
+        throws IOException, InterruptedException {
+        verifyNoRedeliveryWithAutoAck(recoverAsync);
+    }
+
+    public void testNoRedeliveryWithAutoAck()
+        throws IOException, InterruptedException {
+        verifyNoRedeliveryWithAutoAck(recoverSync);
+    }
 }