Support consumer cancellation. Sans documentation atm. And some of the tests will likely fail now and need adjustment

Author: Matthew Sackman

src/com/rabbitmq/client/Channel.java

@@ -180,6 +180,9 @@ public interface Channel extends ShutdownNotifier {
      */
     void setDefaultConsumer(Consumer consumer);
 
+    ConsumerCancellationListener getConsumerCancellationListener();
+    void setConsumerCancellationListener(ConsumerCancellationListener notifier);
+
     /**
      * Request specific "quality of service" settings.
      *

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

@@ -23,6 +23,7 @@
 import com.rabbitmq.client.Command;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.ConsumerCancellationListener;
 import com.rabbitmq.client.Envelope;
 import com.rabbitmq.client.FlowListener;
 import com.rabbitmq.client.GetResponse;
@@ -90,6 +91,8 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
      */
     public volatile ConfirmListener confirmListener = null;
 
+    public volatile ConsumerCancellationListener consumerCancellationListener = null;
+    
     /** Sequence number of next published message requiring confirmation.
      */
     private long nextPublishSeqNo = 0L;
@@ -175,6 +178,15 @@ public void setDefaultConsumer(Consumer consumer) {
         defaultConsumer = consumer;
     }
 
+    public ConsumerCancellationListener getConsumerCancellationListener() {
+        return consumerCancellationListener;
+    }
+
+    public void setConsumerCancellationListener(
+            ConsumerCancellationListener listener) {
+        consumerCancellationListener = listener;
+    }
+
     /**
      * Protected API - sends a ShutdownSignal to all active consumers.
      * @param signal an exception signalling channel shutdown
@@ -339,6 +351,16 @@ public void releaseChannelNumber() {
                 // be handled by whichever RPC continuation invoked Recover,
                 // so return false
                 return false;
+            } else if (method instanceof Basic.Cancel) {
+                ConsumerCancellationListener l = getConsumerCancellationListener();
+                if (l != null) {
+                    try {
+                        l.handleConsumerCancellation(((Basic.Cancel)method).getConsumerTag());
+                    } catch (Throwable ex) {
+                        _connection.getExceptionHandler().handleConsumerCancellationException(this, ex);
+                    }
+                }
+                return true;
             } else {
                 return false;
             }

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

@@ -55,6 +55,12 @@ public void handleConsumerException(Channel channel, Throwable exception,
                                               + " for channel " + channel);
     }
 
+    public void handleConsumerCancellationException(Channel channel,
+            Throwable exception) {
+        handleChannelKiller(channel, exception, "ConsumerCancellationListener.handleConsumerCancellation");
+
+    }
+
     protected void handleChannelKiller(Channel channel,
                                        Throwable exception,
                                        String what)

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

@@ -78,4 +78,6 @@ void handleConsumerException(Channel channel,
                                  Consumer consumer,
                                  String consumerTag,
                                  String methodName);
+
+    void handleConsumerCancellationException(Channel channel, Throwable exception);
 }

test/src/com/rabbitmq/client/test/AMQConnectionTest.java

@@ -197,6 +197,10 @@ public void handleConsumerException(Channel ch,
             fail("handleConsumerException " + consumerTag + " " + methodName + ": " + ex);
         }
 
+        public void handleConsumerCancellationException(Channel channel, Throwable ex) {
+            fail("handleConsmuerCancellationException " + ex);
+        }
+
         public List<Throwable> getHandledExceptions() {
             return _handledExceptions;
         }