Remove whitespace. Also make this far more useful by having it interrupt someone waiting in the queueingconsumer. Note that we don't throw an exception back to the main reader thread if a msg arrives for this consumer after the cancel. However, the logic in ChannelN shows this is impossible anyway. The point is that it's not a channel level or even connection level exception so the main thread should not be upset by this

Author: Matthew Sackman

src/com/rabbitmq/client/Consumer.java

@@ -54,7 +54,7 @@ public interface Consumer {
      * basicCancel: e.g. the queue has been deleted (either by this channel or
      * by any other channel). See handleCancelOk for notification of consumer
      * cancellation due to basicCancel.
-     * 
+     *
      * @throws IOException
      */
     void handleCancel(String consumerTag) throws IOException;

src/com/rabbitmq/client/ConsumerCancelledException.java

@@ -0,0 +1,34 @@
+//  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 Developer of the Original Code is VMware, Inc.
+//  Copyright (c) 2007-2011 VMware, Inc.  All rights reserved.
+//
+
+package com.rabbitmq.client;
+
+import com.rabbitmq.utility.SensibleClone;
+
+public class ConsumerCancelledException extends RuntimeException implements
+        SensibleClone<ConsumerCancelledException> {
+
+    @Override
+    public ConsumerCancelledException sensibleClone() {
+        try {
+            return (ConsumerCancelledException) super.clone();
+        } catch (CloneNotSupportedException e) {
+            // You've got to be kidding me
+            throw new Error(e);
+        }
+    }
+
+}

src/com/rabbitmq/client/QueueingConsumer.java

@@ -64,6 +64,7 @@ public class QueueingConsumer extends DefaultConsumer {
     // When this is non-null the queue is in shutdown mode and nextDelivery should
     // throw a shutdown signal exception.
     private volatile ShutdownSignalException _shutdown;
+    private volatile ConsumerCancelledException _cancelled;
 
     // Marker object used to signal the queue is in shutdown mode.
     // It is only there to wake up consumers. The canonical representation
@@ -86,6 +87,11 @@ public QueueingConsumer(Channel ch, BlockingQueue<Delivery> q) {
         _queue.add(POISON);
     }
 
+    @Override public void handleCancel(String consumerTag) throws IOException {
+        _cancelled = new ConsumerCancelledException();
+        _queue.add(POISON);
+    }
+
     @Override public void handleDelivery(String consumerTag,
                                Envelope envelope,
                                AMQP.BasicProperties properties,
@@ -150,16 +156,19 @@ private void checkShutdown() {
      */
     private Delivery handle(Delivery delivery) {
         if (delivery == POISON ||
-            delivery == null && _shutdown != null) {
+            delivery == null && (_shutdown != null || _cancelled != null)) {
             if (delivery == POISON) {
                 _queue.add(POISON);
-                if (_shutdown == null) {
+                if (_shutdown == null && _cancelled == null) {
                     throw new IllegalStateException(
-                        "POISON in queue, but null _shutdown. " +
+                        "POISON in queue, but null _shutdown and null _cancelled. " +
                         "This should never happen, please report as a BUG");
                 }
             }
-            throw Utility.fixStackTrace(_shutdown);
+            if (null != _shutdown)
+                throw Utility.fixStackTrace(_shutdown);
+            if (null != _cancelled)
+                throw Utility.fixStackTrace(_cancelled);
         }
         return delivery;
     }
@@ -171,7 +180,7 @@ private Delivery handle(Delivery delivery) {
      * @throws ShutdownSignalException if the connection is shut down while waiting
      */
     public Delivery nextDelivery()
-        throws InterruptedException, ShutdownSignalException
+        throws InterruptedException, ShutdownSignalException, ConsumerCancelledException
     {
         return handle(_queue.take());
     }
@@ -184,7 +193,7 @@ public Delivery nextDelivery()
      * @throws ShutdownSignalException if the connection is shut down while waiting
      */
     public Delivery nextDelivery(long timeout)
-        throws InterruptedException, ShutdownSignalException
+        throws InterruptedException, ShutdownSignalException, ConsumerCancelledException
     {
         return handle(_queue.poll(timeout, TimeUnit.MILLISECONDS));
     }

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

@@ -19,24 +19,31 @@
 import java.io.IOException;
 
 import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.ConsumerCancelledException;
 import com.rabbitmq.client.QueueingConsumer;
+import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.client.test.BrokerTestCase;
 
 public class ConsumerCancelNotificiation extends BrokerTestCase {
 
     private final String queue = "cancel_notification_queue";
-    
+
     private final Object lock = new Object();
-    
+
     private boolean notified = false;
 
+    private boolean failed = false;
+
     public void testConsumerCancellationNotification() throws IOException {
+        synchronized (lock) {
+            notified = false;
+        }
         channel.queueDeclare(queue, false, true, false, null);
         Consumer consumer = new QueueingConsumer(channel) {
             @Override
             public void handleCancel(String consumerTag) throws IOException {
                 synchronized (lock) {
-                    notified = !notified;
+                    notified = true;
                     lock.notifyAll();
                 }
             }
@@ -53,4 +60,49 @@ public void handleCancel(String consumerTag) throws IOException {
             assertTrue(notified);
         }
     }
+
+    public void testConsumerCancellationInterruptsQueuingConsumerWait()
+            throws IOException, InterruptedException {
+        synchronized (lock) {
+            notified = false;
+            failed = false;
+        }
+        channel.queueDeclare(queue, false, true, false, null);
+        final QueueingConsumer consumer = new QueueingConsumer(channel);
+        Runnable receiver = new Runnable() {
+
+            @Override
+            public void run() {
+                try {
+                    consumer.nextDelivery();
+                } catch (ConsumerCancelledException e) {
+                    synchronized (lock) {
+                        notified = true;
+                        lock.notifyAll();
+                        return; // avoid fall through to failure
+                    }
+                } catch (ShutdownSignalException e) {
+                } catch (InterruptedException e) {
+                }
+                synchronized (lock) {
+                    failed = true;
+                    lock.notifyAll();
+                }
+            }
+        };
+        Thread t = new Thread(receiver);
+        t.start();
+        channel.basicConsume(queue, consumer);
+        channel.queueDelete(queue);
+        synchronized (lock) {
+            if (!(notified || failed)) {
+                try {
+                    lock.wait();
+                } catch (InterruptedException e) {
+                }
+            }
+            assertTrue(notified);
+        }
+        t.join();
+    }
 }