Merging default to bug21647

Author: Matthew Sackman

src/com/rabbitmq/client/Consumer.java

@@ -49,6 +49,16 @@ public interface Consumer {
      */
     void handleCancelOk(String consumerTag);
 
+    /**
+     * Called when the consumer is cancelled for reasons other than by a
+     * 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;
+
     /**
      * Called to the consumer that either the channel or the undelying connection has been shut down.
      * @param consumerTag the defined consumerTag (either client- or server-generated)

src/com/rabbitmq/client/DefaultConsumer.java

@@ -53,6 +53,14 @@ public void handleCancelOk(String consumerTag) {
         // no work to do
     }
 
+    /**
+     * No-op implementation of {@link Consumer#handleCancel(String)}
+     * @param consumerTag the defined consumer tag (client- or server-generated)
+     */
+    public void handleCancel(String consumerTag) throws IOException {
+        // no work to do
+    }
+
     /**
      * No-op implementation of {@link Consumer#handleShutdownSignal}.
      */

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

@@ -65,6 +65,7 @@ public static Map<String, Object> defaultClientProperties() {
         capabilities.put("publisher_confirms", true);
         capabilities.put("exchange_exchange_bindings", true);
         capabilities.put("basic.nack", true);
+        capabilities.put("consumer_cancel_notify", true);
         return Frame.buildTable(new Object[] {
                 "product", LongStringHelper.asLongString("RabbitMQ"),
                 "version", LongStringHelper.asLongString(ClientVersion.VERSION),

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

@@ -17,10 +17,15 @@
 
 package com.rabbitmq.client.impl;
 
-import com.rabbitmq.client.ConfirmListener;
-import com.rabbitmq.client.AMQP.BasicProperties;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Command;
+import com.rabbitmq.client.ConfirmListener;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.Consumer;
 import com.rabbitmq.client.Envelope;
@@ -30,6 +35,7 @@
 import com.rabbitmq.client.ReturnListener;
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.client.UnexpectedMethodError;
+import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.impl.AMQImpl.Basic;
 import com.rabbitmq.client.impl.AMQImpl.Channel;
 import com.rabbitmq.client.impl.AMQImpl.Confirm;
@@ -38,12 +44,6 @@
 import com.rabbitmq.client.impl.AMQImpl.Tx;
 import com.rabbitmq.utility.Utility;
 
-import java.io.IOException;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.TimeoutException;
-
 
 /**
  * Main interface to AMQP protocol functionality. Public API -
@@ -339,6 +339,24 @@ public void releaseChannelNumber() {
                 // be handled by whichever RPC continuation invoked Recover,
                 // so return false
                 return false;
+            } else if (method instanceof Basic.Cancel) {
+                Basic.Cancel m = (Basic.Cancel)method;
+                Consumer callback = _consumers.remove(m.consumerTag);
+                if (callback == null) {
+                    callback = defaultConsumer;
+                }
+                if (callback != null) {
+                    try {
+                        callback.handleCancel(m.getConsumerTag());
+                    } catch (Throwable ex) {
+                        _connection.getExceptionHandler().handleConsumerException(this,
+                                                                                  ex,
+                                                                                  callback,
+                                                                                  m.getConsumerTag(),
+                                                                                  "handleCancel");
+                    }
+                }
+                return true;
             } else {
                 return false;
             }

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

@@ -0,0 +1,56 @@
+//  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.test.functional;
+
+import java.io.IOException;
+
+import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.QueueingConsumer;
+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;
+
+    public void testConsumerCancellationNotification() throws IOException {
+        channel.queueDeclare(queue, false, true, false, null);
+        Consumer consumer = new QueueingConsumer(channel) {
+            @Override
+            public void handleCancel(String consumerTag) throws IOException {
+                synchronized (lock) {
+                    notified = !notified;
+                    lock.notifyAll();
+                }
+            }
+        };
+        channel.basicConsume(queue, consumer);
+        channel.queueDelete(queue);
+        synchronized (lock) {
+            if (!notified) {
+                try {
+                    lock.wait();
+                } catch (InterruptedException e) {
+                }
+            }
+            assertTrue(notified);
+        }
+    }
+}

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

@@ -54,6 +54,7 @@ public static TestSuite suite() {
         suite.addTestSuite(DefaultExchange.class);
         suite.addTestSuite(UnbindAutoDeleteExchange.class);
         suite.addTestSuite(Confirm.class);
+        suite.addTestSuite(ConsumerCancelNotificiation.class);
         suite.addTestSuite(UnexpectedFrames.class);
         suite.addTestSuite(PerQueueTTL.class);
         suite.addTestSuite(SaslMechanisms.class);