Merge pull request #173 from rabbitmq/rabbitmq-dotnet-client-13

Deprecate QueueingBasicConsumer and reduce its use by other components

Author: michaelklishin

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

@@ -50,11 +50,8 @@ namespace RabbitMQ.Client
     /// </summary>
     /// <remarks>
     /// Note that the "Handle*" methods run in the connection's thread!
-    /// Consider using <see cref="QueueingBasicConsumer"/>,  which uses a
-    /// <see cref="SharedQueue"/> instance to safely pass received messages across
-    /// to user threads, or  RabbitMQ.Client.MessagePatterns.Subscription,
-    ///  which manages resource declaration and binding in addition to providing a
-    /// thread-safe interface.
+    /// Consider using <see cref="EventingBasicConsumer"/>,  which exposes
+    /// events that can be subscribed to consumer messages.
     /// </remarks>
     public class DefaultBasicConsumer : IBasicConsumer
     {

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

@@ -90,6 +90,7 @@ namespace RabbitMQ.Client
     /// }
     /// </code></example>
     /// </remarks>
+    [Obsolete("Deprecated. Use EventingBasicConsumer or a different consumer interface implementation instead")]
     public class QueueingBasicConsumer : DefaultBasicConsumer, IQueueingBasicConsumer
     {
         /// <summary>

projects/client/RabbitMQ.Client/src/client/messagepatterns/Subscription.cs

@@ -40,11 +40,11 @@
 
 using System;
 using System.Collections;
+using System.Collections.Concurrent;
 using System.IO;
 
-#if NETFX_CORE || NET4  // For Windows 8 Store, but could be .NET 4.0 and greater
+using System.Threading;
 using System.Threading.Tasks;
-#endif
 
 using RabbitMQ.Client.Events;
 using RabbitMQ.Client.Exceptions;
@@ -59,7 +59,7 @@ namespace RabbitMQ.Client.MessagePatterns
     ///</para>
     ///<para>
     /// Once created, the Subscription consumes from a queue (using a
-    /// QueueingBasicConsumer). Received deliveries can be retrieved
+    /// EventingBasicConsumer). Received deliveries can be retrieved
     /// by calling Next(), or by using the Subscription as an
     /// IEnumerator in, for example, a foreach loop.
     ///</para>
@@ -76,8 +76,16 @@ namespace RabbitMQ.Client.MessagePatterns
     public class Subscription : ISubscription
     {
         protected readonly object m_eventLock = new object();
-        protected volatile QueueingBasicConsumer m_consumer;
+        protected volatile EventingBasicConsumer m_consumer;
+        private BlockingCollection<BasicDeliverEventArgs> m_queue = 
+            new BlockingCollection<BasicDeliverEventArgs>(new ConcurrentQueue<BasicDeliverEventArgs>());
+
+        private CancellationTokenSource m_queueCts = new CancellationTokenSource();
 
+#if NETFX_CORE || NET4
+        private ConcurrentQueue<TaskCompletionSource<BasicDeliverEventArgs>> m_waiting = 
+            new ConcurrentQueue<TaskCompletionSource<BasicDeliverEventArgs>>();
+#endif
         ///<summary>Creates a new Subscription in "noAck" mode,
         ///consuming from a named queue.</summary>
         public Subscription(IModel model, string queueName)
@@ -92,7 +100,12 @@ public Subscription(IModel model, string queueName, bool noAck)
             Model = model;
             QueueName = queueName;
             NoAck = noAck;
-            m_consumer = new QueueingBasicConsumer(Model);
+            m_consumer = new EventingBasicConsumer(Model);
+#if NETFX_CORE || NET4
+            m_consumer.Received += (sender, args) => QueueAdd(args); 
+#else
+            m_consumer.Received += (sender, args) => m_queue.Add(args); 
+#endif
             ConsumerTag = Model.BasicConsume(QueueName, NoAck, m_consumer);
             m_consumer.ConsumerCancelled += HandleConsumerCancelled;
             LatestEvent = null;
@@ -105,8 +118,9 @@ public Subscription(IModel model, string queueName, bool noAck, string consumerT
             Model = model;
             QueueName = queueName;
             NoAck = noAck;
-            m_consumer = new QueueingBasicConsumer(Model);
+            m_consumer = new EventingBasicConsumer(Model);
             m_consumer.ConsumerCancelled += HandleConsumerCancelled;
+            m_consumer.Received += (sender, args) => m_queue.Add(args);
             ConsumerTag = Model.BasicConsume(QueueName, NoAck, consumerTag, m_consumer);
             LatestEvent = null;
         }
@@ -228,6 +242,17 @@ public void Close()
 
                     ConsumerTag = null;
                 }
+
+                m_queueCts.Cancel(true);
+                m_queue.Dispose();
+                m_queue = null;
+#if NETFX_CORE || NET4
+                var exn = new EndOfStreamException("Subscription closed");
+                foreach (var tsc in m_waiting)
+                {
+                    tsc.TrySetException(exn);
+                }
+#endif
             }
             catch (OperationInterruptedException)
             {
@@ -301,7 +326,7 @@ public BasicDeliverEventArgs Next()
             // Alias the pointer as otherwise it may change out
             // from under us by the operation of Close() from
             // another thread.
-            QueueingBasicConsumer consumer = m_consumer;
+            EventingBasicConsumer consumer = m_consumer;
             try
             {
                 if (consumer == null || Model.IsClosed)
@@ -310,7 +335,7 @@ public BasicDeliverEventArgs Next()
                 }
                 else
                 {
-                    BasicDeliverEventArgs bdea = consumer.Queue.Dequeue();
+                    BasicDeliverEventArgs bdea = m_queue.Take(m_queueCts.Token);
                     MutateLatestEvent(bdea);
                 }
             }
@@ -322,31 +347,49 @@ public BasicDeliverEventArgs Next()
         }
 
 #if NETFX_CORE || NET4
-        public async Task<BasicDeliverEventArgs> NextAsync() {
-            try {
+        public Task<BasicDeliverEventArgs> NextAsync() 
+        {
+            try 
+            {
                 // Alias the pointer as otherwise it may change out
                 // from under us by the operation of Close() from
                 // another thread.
-                QueueingBasicConsumer consumer = m_consumer;
-                if (consumer == null) {
+                var queue = m_queue;
+                if (queue == null || Model.IsClosed) 
+                {
                     // Closed!
                     MutateLatestEvent(null);
                 }
-                else {
-                    MutateLatestEvent(await consumer.Queue.DequeueAsync());
+                else 
+                {
+                    BasicDeliverEventArgs evt = null;
+                    if(queue.TryTake(out evt))
+                    {
+                        MutateLatestEvent(evt);
+                    }
+                    else
+                    {
+                        var tcs = new TaskCompletionSource<BasicDeliverEventArgs>();
+                        m_waiting.Enqueue(tcs);
+                        return tcs.Task;
+                    }
                 }
             }
-            catch (AggregateException ex) {
+            catch (AggregateException ex) 
+            {
                 // since tasks wrap exceptions as AggregateException, 
                 // reach in and check if the EndOfStream exception is what happened
-                if (ex.InnerException is EndOfStreamException) {
+                if (ex.InnerException is EndOfStreamException) 
+                {
                     MutateLatestEvent(null);
                 }
             }
-            catch (EndOfStreamException) {
+            catch (EndOfStreamException) 
+            {
                 MutateLatestEvent(null);
             }
-            return LatestEvent;
+
+            return Task.FromResult(LatestEvent);
         }
 #endif
 
@@ -401,7 +444,7 @@ public bool Next(int millisecondsTimeout, out BasicDeliverEventArgs result)
                 // Alias the pointer as otherwise it may change out
                 // from under us by the operation of Close() from
                 // another thread.
-                QueueingBasicConsumer consumer = m_consumer;
+                var consumer = m_consumer;
                 if (consumer == null || Model.IsClosed)
                 {
                     MutateLatestEvent(null);
@@ -411,7 +454,7 @@ public bool Next(int millisecondsTimeout, out BasicDeliverEventArgs result)
                 else
                 {
                     BasicDeliverEventArgs qValue;
-                    if (!consumer.Queue.Dequeue(millisecondsTimeout, out qValue))
+                    if (!m_queue.TryTake(out qValue, millisecondsTimeout))
                     {
                         result = null;
                         return false;
@@ -483,5 +526,22 @@ private void HandleConsumerCancelled(object sender, ConsumerEventArgs e)
                 MutateLatestEvent(null);
             }
         }
+
+#if NETFX_CORE || NET4
+        private void QueueAdd(BasicDeliverEventArgs args)
+        {
+            //NB: as long as there are async awaiters sync callers will never be served
+            //this is not ideal but consistent with how SharedQueue behaves
+            TaskCompletionSource<BasicDeliverEventArgs> tsc;
+            if(m_waiting.TryDequeue(out tsc) && tsc.TrySetResult(args))
+            {
+                return;
+            }
+            else
+            {
+                m_queue.Add(args);
+            }
+        }
+#endif
     }
 }

projects/wcf/RabbitMQ.ServiceModel/src/serviceModel/RabbitMQInputChannel.cs

@@ -39,6 +39,7 @@
 //---------------------------------------------------------------------------
 
 using System;
+using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.IO;
 using System.ServiceModel;
@@ -55,7 +56,9 @@ internal sealed class RabbitMQInputChannel : RabbitMQInputChannelBase
         private RabbitMQTransportBindingElement m_bindingElement;
         private MessageEncoder m_encoder;
         private IModel m_model;
-        private QueueingBasicConsumer m_messageQueue;
+        private EventingBasicConsumer m_consumer;
+        private BlockingCollection<BasicDeliverEventArgs> m_queue = 
+            new BlockingCollection<BasicDeliverEventArgs>(new ConcurrentQueue<BasicDeliverEventArgs>());
 
         public RabbitMQInputChannel(BindingContext context, IModel model, EndpointAddress address)
             : base(context, address)
@@ -67,21 +70,21 @@ public RabbitMQInputChannel(BindingContext context, IModel model, EndpointAddres
                 m_encoder = encoderElem.CreateMessageEncoderFactory().Encoder;
             }
             m_model = model;
-            m_messageQueue = null;
+            m_consumer = null;
         }
 
 
-        public override Message Receive(TimeSpan timeout)
+        public override Message Receive(TimeSpan timeout) //TODO: timeout isn't used
         {
             try
             {
-                BasicDeliverEventArgs msg = m_messageQueue.Queue.Dequeue() as BasicDeliverEventArgs;
+                BasicDeliverEventArgs msg = m_queue.Take();
 #if VERBOSE
                 DebugHelper.Start();
 #endif
                 Message result = m_encoder.ReadMessage(new MemoryStream(msg.Body), (int)m_bindingElement.MaxReceivedMessageSize);
                 result.Headers.To = base.LocalAddress.Uri;
-                m_messageQueue.Model.BasicAck(msg.DeliveryTag, false);
+                m_consumer.Model.BasicAck(msg.DeliveryTag, false);
 #if VERBOSE
                 DebugHelper.Stop(" #### Message.Receive {{\n\tAction={2}, \n\tBytes={1}, \n\tTime={0}ms}}.",
                         msg.Body.Length,
@@ -91,7 +94,7 @@ public override Message Receive(TimeSpan timeout)
             }
             catch (EndOfStreamException)
             {
-                if (m_messageQueue== null || m_messageQueue.ShutdownReason != null && m_messageQueue.ShutdownReason.ReplyCode != Constants.ReplySuccess)
+                if (m_consumer== null || m_consumer.ShutdownReason != null && m_consumer.ShutdownReason.ReplyCode != Constants.ReplySuccess)
                 {
                     OnFaulted();
                 }
@@ -124,9 +127,9 @@ public override void Close(TimeSpan timeout)
 #if VERBOSE
             DebugHelper.Start();
 #endif
-            if (m_messageQueue != null) {
-                m_model.BasicCancel(m_messageQueue.ConsumerTag);
-                m_messageQueue = null;
+            if (m_consumer != null) {
+                m_model.BasicCancel(m_consumer.ConsumerTag);
+                m_consumer = null;
             }
 #if VERBOSE
             DebugHelper.Stop(" ## In.Channel.Close {{\n\tAddress={1}, \n\tTime={0}ms}}.", LocalAddress.Uri.PathAndQuery);
@@ -148,8 +151,9 @@ public override void Open(TimeSpan timeout)
             m_model.QueueBind(queue, Exchange, base.LocalAddress.Uri.PathAndQuery, null);
 
             //Listen to the queue
-            m_messageQueue = new QueueingBasicConsumer(m_model);
-            m_model.BasicConsume(queue, false, m_messageQueue);
+            m_consumer = new EventingBasicConsumer(m_model);
+            m_consumer.Received += (sender, args) => m_queue.Add(args);
+            m_model.BasicConsume(queue, false, m_consumer);
 
 #if VERBOSE
             DebugHelper.Stop(" ## In.Channel.Open {{\n\tAddress={1}, \n\tTime={0}ms}}.", LocalAddress.Uri.PathAndQuery);