Merge pull request #759 from rabbitmq/rabbitmq-dotnet-client-341-BasicCancelNoWait

Expose nowait version of BasicCancel

Author: lukebakken

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

@@ -193,6 +193,12 @@ public interface IModel : IDisposable
         [AmqpMethodDoNotImplement(null)]
         void BasicCancel(string consumerTag);
 
+        /// <summary>
+        /// Same as BasicCancel but sets nowait to true and returns void (as there
+        /// will be no response from the server).
+        /// </summary>
+        void BasicCancelNoWait(string consumerTag);
+
         /// <summary>Start a Basic content-class consumer.</summary>
         [AmqpMethodDoNotImplement(null)]
         string BasicConsume(

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

@@ -758,6 +758,16 @@ public void BasicCancel(string consumerTag)
             _delegate.BasicCancel(consumerTag);
         }
 
+        public void BasicCancelNoWait(string consumerTag)
+        {
+            RecordedConsumer cons = _connection.DeleteRecordedConsumer(consumerTag);
+            if (cons != null)
+            {
+                _connection.MaybeDeleteRecordedAutoDeleteQueue(cons.Queue);
+            }
+            _delegate.BasicCancelNoWait(consumerTag);
+        }
+
         public string BasicConsume(
             string queue,
             bool autoAck,

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

@@ -1013,6 +1013,7 @@ public void BasicCancel(string consumerTag)
                 _Private_BasicCancel(consumerTag, false);
                 k.GetReply(ContinuationTimeout);
             }
+
             lock (m_consumers)
             {
                 m_consumers.Remove(consumerTag);
@@ -1021,6 +1022,16 @@ public void BasicCancel(string consumerTag)
             ModelShutdown -= k.m_consumer.HandleModelShutdown;
         }
 
+        public void BasicCancelNoWait(string consumerTag)
+        {
+            _Private_BasicCancel(consumerTag, true);
+
+            lock (m_consumers)
+            {
+                m_consumers.Remove(consumerTag);
+            }
+        }
+
         public string BasicConsume(string queue,
             bool autoAck,
             string consumerTag,
@@ -1224,7 +1235,7 @@ public void ExchangeDelete(string exchange,
         public void ExchangeDeleteNoWait(string exchange,
             bool ifUnused)
         {
-            _Private_ExchangeDelete(exchange, ifUnused, false);
+            _Private_ExchangeDelete(exchange, ifUnused, true);
         }
 
         public void ExchangeUnbind(string destination,

projects/client/Unit/src/unit/APIApproval.Approve.approved.txt

@@ -369,6 +369,7 @@ namespace RabbitMQ.Client
         void Abort(ushort replyCode, string replyText);
         void BasicAck(ulong deliveryTag, bool multiple);
         void BasicCancel(string consumerTag);
+        void BasicCancelNoWait(string consumerTag);
         string BasicConsume(string queue, bool autoAck, string consumerTag, bool noLocal, bool exclusive, System.Collections.Generic.IDictionary<string, object> arguments, RabbitMQ.Client.IBasicConsumer consumer);
         RabbitMQ.Client.BasicGetResult BasicGet(string queue, bool autoAck);
         void BasicNack(ulong deliveryTag, bool multiple, bool requeue);

projects/client/Unit/src/unit/TestAsyncConsumer.cs

@@ -51,7 +51,6 @@ namespace RabbitMQ.Client.Unit
     [TestFixture]
     public class TestAsyncConsumer
     {
-
         [Test]
         public void TestBasicRoundtrip()
         {
@@ -82,6 +81,38 @@ public void TestBasicRoundtrip()
             }
         }
 
+        [Test]
+        public void TestBasicRoundtripNoWait()
+        {
+            var cf = new ConnectionFactory{ DispatchConsumersAsync = true };
+            using (IConnection c = cf.CreateConnection())
+            {
+                using (IModel m = c.CreateModel())
+                {
+                    QueueDeclareOk q = m.QueueDeclare();
+                    IBasicProperties bp = m.CreateBasicProperties();
+                    byte[] body = System.Text.Encoding.UTF8.GetBytes("async-hi");
+                    m.BasicPublish("", q.QueueName, bp, body);
+                    var consumer = new AsyncEventingBasicConsumer(m);
+                    var are = new AutoResetEvent(false);
+                    consumer.Received += async (o, a) =>
+                        {
+                            are.Set();
+                            await Task.Yield();
+                        };
+                    string tag = m.BasicConsume(q.QueueName, true, consumer);
+                    // ensure we get a delivery
+                    bool waitRes = are.WaitOne(2000);
+                    Assert.IsTrue(waitRes);
+                    // unsubscribe and ensure no further deliveries
+                    m.BasicCancelNoWait(tag);
+                    m.BasicPublish("", q.QueueName, bp, body);
+                    bool waitResFalse = are.WaitOne(2000);
+                    Assert.IsFalse(waitResFalse);
+                }
+            }
+        }
+
         [Test]
         public void NonAsyncConsumerShouldThrowInvalidOperationException()
         {
@@ -98,4 +129,4 @@ public void NonAsyncConsumerShouldThrowInvalidOperationException()
             }
         }
     }
-}
+}