Added support for consumer_cancel_notify to .Net client

Author: Matthew Sackman

docs/specs/amqp0-9-1.stripped.xml

@@ -384,12 +384,14 @@ THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
     </method>
     <method name="cancel" synchronous="1" index="30">
       <chassis name="server" implement="MUST"/>
+      <chassis name="client" implement="SHOULD"/>
       <response name="cancel-ok"/>
       <field name="consumer-tag" domain="consumer-tag"/>
       <field name="no-wait" domain="no-wait"/>
     </method>
     <method name="cancel-ok" synchronous="1" index="31">
       <chassis name="client" implement="MUST"/>
+      <chassis name="server" implement="MAY"/>
       <field name="consumer-tag" domain="consumer-tag"/>
     </method>
     <method name="publish" content="1" index="40">

docs/specs/amqp0-9-1.xml

@@ -7,7 +7,8 @@
      exchange.bind and exchange.bind-ok,
      exchange.unbind and exchange.unbind-ok,
      basic.nack
-     and the ability for the Server to send basic.ack and basic.nack to the client.
+     and the ability for the Server to send basic.ack, basic.nack and
+      basic.cancel to the client.
 -->
 
 <!--
@@ -154,9 +155,8 @@
 <amqp major = "0" minor = "9" revision = "1"
     port = "5672" comment = "AMQ Protocol version 0-9-1">
   <!--
-      ======================================================
-      ==       CONSTANTS
-      ======================================================
+      ====================================================== ==
+      CONSTANTS ======================================================
   -->
   <!-- Frame types -->
   <constant name = "frame-method"     value = "1" />
@@ -2533,6 +2533,13 @@
         messages, but it does mean the server will not send any more messages for
         that consumer. The client may receive an arbitrary number of messages in
         between sending the cancel method and receiving the cancel-ok reply.
+
+        It is also sent from the server to the client in the event of
+        the consumer being unexpectedly cancelled (i.e. cancelled for
+        any reason other than the server receiving the corresponding
+        basic.cancel from the client). This allows clients to be
+        notified of the loss of consumers due to events such as queue
+        deletion.
       </doc>
 
       <rule name = "01">
@@ -2546,6 +2553,7 @@
       </rule>
 
       <chassis name = "server" implement = "MUST" />
+      <chassis name = "client" implement = "SHOULD" />
       <response name = "cancel-ok" />
 
       <field name = "consumer-tag" domain = "consumer-tag" />
@@ -2557,6 +2565,7 @@
         This method confirms that the cancellation was completed.
       </doc>
       <chassis name = "client" implement = "MUST" />
+      <chassis name = "server" implement = "MAY" />
       <field name = "consumer-tag" domain = "consumer-tag" />
     </method>
 

projects/client/RabbitMQ.Client/src/client/api/DefaultBasicConsumer.cs

@@ -131,6 +131,12 @@ public virtual void HandleBasicCancelOk(string consumerTag)
             OnCancel();
         }
 
+        ///<summary>Default implementation - calls OnCancel().</summary>
+        public virtual void HandleBasicCancel(string consumerTag)
+        {
+            OnCancel();
+        }
+        
         ///<summary>Default implementation - sets ShutdownReason and
         ///calls OnCancel().</summary>
         public virtual void HandleModelShutdown(IModel model, ShutdownEventArgs reason)

projects/client/RabbitMQ.Client/src/client/api/IBasicConsumer.cs

@@ -69,6 +69,14 @@ public interface IBasicConsumer
         ///<summary>Called upon successful deregistration of the
         ///consumer from the broker.</summary>
         void HandleBasicCancelOk(string consumerTag);
+        
+        /// <summary>
+        /// 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.
+        /// </summary>
+        void HandleBasicCancel(string consumerTag);
 
         ///<summary>Called when the model shuts down.</summary>
         void HandleModelShutdown(IModel model, ShutdownEventArgs reason);

projects/client/RabbitMQ.Client/src/client/api/IModel.cs

@@ -801,6 +801,8 @@ void HandleBasicDeliver(string consumerTag,
                                 [AmqpContentBodyMapping]
                                 byte[] body);
 
+        void HandleBasicCancel(string consumerTag, bool nowait);
+
         ///<summary>Handle incoming Basic.Return methods. Signals a
         ///BasicReturnEvent.</summary>
         void HandleBasicReturn(ushort replyCode,

projects/client/RabbitMQ.Client/src/client/impl/ModelBase.cs

@@ -556,6 +556,29 @@ public void HandleBasicDeliver(string consumerTag,
             }
         }
 
+        public void HandleBasicCancel(string consumerTag, bool nowait)
+        {
+            IBasicConsumer consumer;
+            lock (m_consumers)
+            {
+                consumer = (IBasicConsumer)m_consumers[consumerTag];
+                m_consumers.Remove(consumerTag);
+            }
+            if (consumer == null)
+            {
+                consumer = DefaultConsumer;
+            }
+
+            try {
+                consumer.HandleBasicCancel(consumerTag);
+            } catch (Exception e) {
+                CallbackExceptionEventArgs args = new CallbackExceptionEventArgs(e);
+                args.Detail["consumer"] = consumer;
+                args.Detail["context"] = "HandleBasicCancel";
+                OnCallbackException(args);
+            }
+        }
+
         public void HandleBasicReturn(ushort replyCode,
                                       string replyText,
                                       string exchange,

projects/client/RabbitMQ.Client/src/client/impl/v0_9_1/ProtocolBase.cs

@@ -51,6 +51,7 @@ public ProtocolBase() {
             Capabilities["publisher_confirms"] = true;
             Capabilities["exchange_exchange_bindings"] = true;
             Capabilities["basic.nack"] = true;
+            Capabilities["consumer_cancel_notify"] = true;
         }
 
         public override IFrameHandler CreateFrameHandler(AmqpTcpEndpoint endpoint) {