Merge bug26097 into default

Author: Tim Watson

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

@@ -80,6 +80,7 @@ public class Subscription: IEnumerable, IEnumerator, IDisposable {
         protected bool m_noAck;
 
         protected readonly object m_consumerLock = new object();
+        protected readonly object m_eventLock = new object();
         protected volatile QueueingBasicConsumer m_consumer;
         protected string m_consumerTag;
 
@@ -156,7 +157,11 @@ public void Close()
                 }
 
                 if (shouldCancelConsumer) {
-                    m_model.BasicCancel(m_consumerTag);
+                    if(m_model.IsOpen)
+                    {
+                        m_model.BasicCancel(m_consumerTag);
+                    }
+                    
                     m_consumerTag = null;
                 }
             } catch (OperationInterruptedException) {
@@ -169,34 +174,73 @@ public void Close()
         ///null.</summary>
         public void Ack()
         {
-            if (m_latestEvent != null) {
-                Ack(m_latestEvent);
-            }
+            Ack(m_latestEvent);
         }
 
         ///<summary>If we are not in "noAck" mode, calls
-        ///IModel.BasicAck with the delivery-tag from the passed in
-        ///event; otherwise, sends nothing to the server. In both
-        ///cases, if the passed-in event is the same as LatestEvent
-        ///(by pointer comparison), sets LatestEvent to
-        ///null.</summary>
+        ///IModel.BasicAck with the delivery-tag from <paramref name="evt"/>;
+        ///otherwise, sends nothing to the server. if <paramref name="evt"/> is the same as LatestEvent
+        ///by pointer comparison, sets LatestEvent to null.
+        ///</summary>
         ///<remarks>
-        /// Make sure that this method is only called with events that
-        /// originated from this Subscription - other usage will have
-        /// unpredictable results.
+        ///Passing an event that did not originate with this Subscription's
+        /// channel, will lead to unpredictable behaviour
         ///</remarks>
         public void Ack(BasicDeliverEventArgs evt)
         {
             if (evt == null) {
                 return;
             }
 
-            if (!m_noAck) {
+            if (!m_noAck && m_model.IsOpen) {
                 m_model.BasicAck(evt.DeliveryTag, false);
             }
 
             if (evt == m_latestEvent) {
-                m_latestEvent = null;
+                MutateLatestEvent(null);
+            }
+        }
+
+        ///<summary>If LatestEvent is non-null, passes it to
+        ///Nack(BasicDeliverEventArgs, false, requeue). Causes LatestEvent to become
+        ///null.</summary>
+        public void Nack(bool requeue)
+        {
+            Nack(m_latestEvent, false, requeue);
+        }
+
+
+        ///<summary>If LatestEvent is non-null, passes it to
+        ///Nack(BasicDeliverEventArgs, multiple, requeue). Causes LatestEvent to become
+        ///null.</summary>
+        public void Nack(bool multiple, bool requeue)
+        {
+            Nack(m_latestEvent, multiple, requeue);
+        }
+
+        ///<summary>If we are not in "noAck" mode, calls
+        ///IModel.BasicNack with the delivery-tag from <paramref name="evt"/>;
+        ///otherwise, sends nothing to the server. if <paramref name="evt"/> is the same as LatestEvent
+        ///by pointer comparison, sets LatestEvent to null.
+        ///</summary>
+        ///<remarks>
+        ///Passing an event that did not originate with this Subscription's
+        /// channel, will lead to unpredictable behaviour
+        ///</remarks>
+        public void Nack(BasicDeliverEventArgs evt,
+                         bool multiple,
+                         bool requeue)
+        {
+            if (evt == null) {
+                return;
+            }
+
+            if (!m_noAck && m_model.IsOpen) {
+                m_model.BasicNack(evt.DeliveryTag, multiple, requeue);
+            }
+
+            if (evt == m_latestEvent) {
+                MutateLatestEvent(null);
             }
         }
 
@@ -220,19 +264,19 @@ public void Ack(BasicDeliverEventArgs evt)
         ///</remarks>
         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;
             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 || m_model.IsClosed) {
-                    // Closed!
-                    m_latestEvent = null;
+                    MutateLatestEvent(null);
                 } else {
-                    m_latestEvent = (BasicDeliverEventArgs) consumer.Queue.Dequeue();
+                    BasicDeliverEventArgs bdea = (BasicDeliverEventArgs) consumer.Queue.Dequeue();
+                    MutateLatestEvent(bdea);
                 }
             } catch (EndOfStreamException) {
-                m_latestEvent = null;
+                MutateLatestEvent(null);
             }
             return m_latestEvent;
         }
@@ -289,8 +333,7 @@ public bool Next(int millisecondsTimeout, out BasicDeliverEventArgs result)
                 // another thread.
                 QueueingBasicConsumer consumer = m_consumer;
                 if (consumer == null || m_model.IsClosed) {
-                    // Closed!
-                    m_latestEvent = null;
+                    MutateLatestEvent(null);
                     result = null;
                     return false;
                 } else {
@@ -299,10 +342,10 @@ public bool Next(int millisecondsTimeout, out BasicDeliverEventArgs result)
                         result = null;
                         return false;
                     }
-                    m_latestEvent = qValue;
+                    MutateLatestEvent(qValue);
                 }
             } catch (EndOfStreamException) {
-                m_latestEvent = null;
+                MutateLatestEvent(null);
             }
             result = m_latestEvent;
             return true;
@@ -369,5 +412,13 @@ void IDisposable.Dispose()
         {
             Close();
         }
+
+        protected void MutateLatestEvent(BasicDeliverEventArgs value)
+        {
+            lock(m_eventLock)
+            {
+                m_latestEvent = value;
+            }
+        }
     }
 }

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

@@ -43,14 +43,18 @@
 using System;
 using System.Text;
 using System.Threading;
-using System.Diagnostics;
+using System.Collections.Generic;
+using System.Timers;
 
+using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Events;
 using RabbitMQ.Client.MessagePatterns;
 
-namespace RabbitMQ.Client.Unit {
+namespace RabbitMQ.Client.Unit
+{
     [TestFixture]
-    public class TestMessagePatternsSubscription : IntegrationFixture {
+    public class TestMessagePatternsSubscription : IntegrationFixture
+    {
         UTF8Encoding enc = new UTF8Encoding();
 
         [Test]
@@ -75,5 +79,122 @@ public void TestChannelClosureIsObservableOnSubscription()
             BasicDeliverEventArgs r3;
             Assert.IsFalse(sub.Next(100, out r3));
         }
+
+        [Test]
+        public void TestSubscriptionAck()
+        {
+            TestSubscriptionAction((s) => s.Ack());
+        }
+
+        [Test]
+        public void TestSubscriptionNack()
+        {
+            TestSubscriptionAction((s) => s.Nack(false, false));
+        }
+
+        [Test]
+        public void TestConcurrentIterationAndAck()
+        {
+            TestConcurrentIterationWithDrainer((s) => s.Ack());
+        }
+
+        [Test]
+        public void TestConcurrentIterationAndNack()
+        {
+            TestConcurrentIterationWithDrainer((s) => s.Nack(false, false));
+        }
+
+        protected void TestConcurrentIterationWithDrainer(SubscriptionAction act)
+        {
+            IDictionary<string, object> args = new Dictionary<string, object>
+            {
+                {"x-message-ttl", 5000}
+            };
+            string q = Model.QueueDeclare("", false, true, false, args);
+            Subscription sub = new Subscription(Model, q, false);
+
+            PreparedQueue(q);
+
+            List<Thread> ts = new List<Thread>();
+            for (int i = 0; i < 50; i++)
+            {
+                SubscriptionDrainer drainer = new SubscriptionDrainer(sub, act);
+                Thread t = new Thread(drainer.Drain);
+                ts.Add(t);
+                t.Start();
+            }
+
+            foreach(Thread t in ts)
+            {
+                t.Join();
+            }
+        }
+
+        private void TestSubscriptionAction(SubscriptionAction action)
+        {
+            Model.BasicQos(0, 1, false);
+            string q = Model.QueueDeclare();
+            Subscription sub = new Subscription(Model, q, false);
+
+            Model.BasicPublish("", q, null, enc.GetBytes("a message"));
+            BasicDeliverEventArgs res = sub.Next();
+            Assert.IsNotNull(res);
+            action(sub);
+            QueueDeclareOk ok = Model.QueueDeclarePassive(q);
+            Assert.AreEqual(0, ok.MessageCount);
+        }
+
+        protected delegate void SubscriptionAction(Subscription s);
+
+        protected class SubscriptionDrainer
+        {
+            protected Subscription m_subscription;
+            private SubscriptionAction PostProcess { get; set; }
+
+            public SubscriptionDrainer(Subscription sub, SubscriptionAction op)
+            {
+                m_subscription = sub;
+                PostProcess = op;
+            }
+
+            public void Drain()
+            {
+                #pragma warning disable 0168
+                try
+                {
+                    for(int i = 0; i < 100; i++)
+                    {
+                        BasicDeliverEventArgs ea = m_subscription.Next();
+                        if(ea != null)
+                        {
+                            Assert.That(ea, Is.TypeOf(typeof(BasicDeliverEventArgs)));
+                            this.PostProcess(m_subscription);
+                        }
+                        else
+                        {
+                            break;
+                        }
+                    }
+                }
+                catch (AlreadyClosedException ace)
+                {
+                    // expected
+                }
+                finally
+                {
+                    m_subscription.Close();
+                }
+                #pragma warning restore
+
+            }
+        }
+
+        private void PreparedQueue(string q)
+        {
+            for (int i = 0; i < 1024; i++)
+            {
+                Model.BasicPublish("", q, null, enc.GetBytes("a message"));
+            }
+        }
     }
 }