Merging bug 22587 into default

Author: Matthew Sackman

src/com/rabbitmq/client/Channel.java

@@ -147,6 +147,40 @@ public interface Channel extends ShutdownNotifier {
      */
     void setFlowListener(FlowListener listener);
 
+    /**
+     * Get the current default consumer. @see setDefaultConsumer for rationale.
+     * @return an interface to the current default consumer.
+     */
+    Consumer getDefaultConsumer();
+
+    /**
+     * Set the current default consumer.
+     *
+     * Under certain circumstances it is possible for a channel to receive a
+     * message delivery which does not match any consumer which is currently
+     * set up via basicConsume(). This will occur after the following sequence
+     * of events:
+     *
+     * ctag = basicConsume(queue, consumer); // i.e. with explicit acks
+     * // some deliveries take place but are not acked
+     * basicCancel(ctag);
+     * basicRecover(false);
+     *
+     * Since requeue is specified to be false in the basicRecover, the spec
+     * states that the message must be redelivered to "the original recipient"
+     * - i.e. the same channel / consumer-tag. But the consumer is no longer
+     * active.
+     *
+     * In these circumstances, you can register a default consumer to handle
+     * such deliveries. If no default consumer is registered an
+     * IllegalStateException will be thrown when such a delivery arrives.
+     *
+     * Most people will not need to use this.
+     *
+     * @param consumer the consumer to use, or null indicating "don't use one".
+     */
+    void setDefaultConsumer(Consumer consumer);
+
     /**
      * Request specific "quality of service" settings.
      *

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

@@ -98,6 +98,11 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
      */
     public volatile FlowListener flowListener = null;
 
+    /** Reference to the currently-active default consumer, or null if there is
+     *  none.
+     */
+    public volatile Consumer defaultConsumer = null;
+
     /**
      * Construct a new channel on the given connection with the given
      * channel number. Usually not called directly - call
@@ -148,6 +153,19 @@ public void setFlowListener(FlowListener listener) {
         flowListener = listener;
     }
 
+    /** Returns the current default consumer. */
+    public Consumer getDefaultConsumer() {
+        return defaultConsumer;
+    }
+
+    /**
+     * Sets the current default consumer.
+     * A null argument is interpreted to mean "do not use a default consumer".
+     */
+    public void setDefaultConsumer(Consumer consumer) {
+        defaultConsumer = consumer;
+    }
+
     /**
      * Protected API - sends a ShutdownSignal to all active consumers.
      * @param signal an exception signalling channel shutdown
@@ -228,8 +246,17 @@ public void releaseChannelNumber() {
 
                 Consumer callback = _consumers.get(m.consumerTag);
                 if (callback == null) {
-                    // FIXME: what to do when we get such an unsolicited delivery?
-                    throw new UnsupportedOperationException("FIXME unsolicited delivery");
+                    if (defaultConsumer == null) {
+                        // No handler set. We should blow up as this message
+                        // needs acking, just dropping it is not enough. See bug
+                        // 22587 for discussion.
+                        throw new IllegalStateException("Unsolicited delivery -" +
+                                " see Channel.setDefaultConsumer to handle this" +
+                                " case.");
+                    }
+                    else {
+                        callback = defaultConsumer;
+                    }
                 }
 
                 Envelope envelope = new Envelope(m.deliveryTag,

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

@@ -62,6 +62,7 @@ public static TestSuite suite() {
         suite.addTestSuite(InvalidAcksTx.class);
         suite.addTestSuite(BindToDefaultExchange.class);
         suite.addTestSuite(UnbindAutoDeleteExchange.class);
+        suite.addTestSuite(RecoverAfterCancel.class);
         return suite;
     }
 }

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

@@ -0,0 +1,43 @@
+package com.rabbitmq.client.test.functional;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.QueueingConsumer;
+import com.rabbitmq.client.test.BrokerTestCase;
+
+import java.io.IOException;
+
+/**
+ * After a basic.cancel you can invoke basic.recover{requeue=false} and
+ * get back delivery(s) with now-obsolete ctags.
+ *
+ * See bug 22587.
+ */
+public class RecoverAfterCancel extends BrokerTestCase {
+    String queue;
+
+    public void createResources() throws IOException {
+        AMQP.Queue.DeclareOk ok = channel.queueDeclare();
+        queue = ok.getQueue();
+    }
+
+    public void testRecoverAfterCancel() throws IOException, InterruptedException {
+        basicPublishVolatile(queue);
+        QueueingConsumer consumer = new QueueingConsumer(channel);
+        QueueingConsumer defaultConsumer = new QueueingConsumer(channel);
+        channel.setDefaultConsumer(defaultConsumer);
+        String ctag = channel.basicConsume(queue, consumer);
+        QueueingConsumer.Delivery del = consumer.nextDelivery();
+        channel.basicCancel(ctag);
+        channel.basicRecover(false);
+
+        // The server will now redeliver us the first message again, with the
+        // same ctag, but we're not set up to handle it with a standard
+        // consumer - it should end up with the default one.
+
+        QueueingConsumer.Delivery del2 = defaultConsumer.nextDelivery();
+
+        assertEquals(new String(del.getBody()), new String(del2.getBody()));
+        assertFalse(del.getEnvelope().isRedeliver());
+        assertTrue(del2.getEnvelope().isRedeliver());
+    }
+}