merge bug21465 into default

Author: tonyg

src/client/api/QueueingBasicConsumer.cs

@@ -76,8 +76,8 @@ namespace RabbitMQ.Client
     /// When the consumer is closed, through BasicCancel or through
     /// the shutdown of the underlying IModel or IConnection, the
     /// SharedQueue's Close() method is called, which causes any
-    /// threads blocked on the queue's Enqueue() or Dequeue()
-    /// operations to throw EndOfStreamException (see the comment for
+    /// Enqueue() operations, and Dequeue() operations when the queue
+    /// is empty, to throw EndOfStreamException (see the comment for
     /// SharedQueue.Close()).
     ///</para>
     ///<para>

src/unit/TestSharedQueue.cs

@@ -65,24 +65,52 @@
 
 [TestFixture]
 public class TestSharedQueue {
+
+    public delegate void Thunk();
+
     public class DelayedEnqueuer
     {
         public SharedQueue m_q;
         public int m_delayMs;
         public object m_v;
-        public void Run()
+	public DelayedEnqueuer(SharedQueue q, int delayMs, object v) {
+	    m_q = q;
+	    m_delayMs = delayMs;
+	    m_v = v;
+	}
+        public void EnqueueValue()
         {
             Thread.Sleep(m_delayMs);
             m_q.Enqueue(m_v);
         }
+	public void Dequeue()
+	{
+	    m_q.Dequeue();
+	}
+	public void DequeueNoWaitZero()
+	{
+	    m_q.DequeueNoWait(0);
+	}
+	public void DequeueAfterOneIntoV()
+	{
+	    m_q.Dequeue(1, out m_v);
+	}
+	public void BackgroundEofExpectingDequeue()
+	{
+	    ExpectEof(new Thunk(this.Dequeue));
+	}
     }
 
     public static void EnqueueAfter(int delayMs, SharedQueue q, object v) {
-        DelayedEnqueuer de = new DelayedEnqueuer();
-        de.m_q = q;
-        de.m_delayMs = delayMs;
-        de.m_v = v;
-        new Thread(new ThreadStart(de.Run)).Start();
+        DelayedEnqueuer de = new DelayedEnqueuer(q, delayMs, v);
+        new Thread(new ThreadStart(de.EnqueueValue)).Start();
+    }
+
+    public static void ExpectEof(Thunk thunk) {
+        try {
+            thunk();
+            Assert.Fail("expected System.IO.EndOfStreamException");
+        } catch (System.IO.EndOfStreamException) {}
     }
 
     public DateTime m_startTime;
@@ -248,4 +276,44 @@ public void TestDoublePoll() {
         Assert.IsTrue(r);
         Assert.AreEqual(123, v);
     }
+
+    [Test]
+    public void TestCloseWhenEmpty() {
+	DelayedEnqueuer de = new DelayedEnqueuer(new SharedQueue(), 0, 1);
+        de.m_q.Close();
+        ExpectEof(new Thunk(de.EnqueueValue));
+        ExpectEof(new Thunk(de.Dequeue));
+        ExpectEof(new Thunk(de.DequeueNoWaitZero));
+        ExpectEof(new Thunk(de.DequeueAfterOneIntoV));
+    }
+
+    [Test]
+    public void TestCloseWhenFull() {
+        SharedQueue q = new SharedQueue();
+        object v; 
+        q.Enqueue(1);
+        q.Enqueue(2);
+        q.Enqueue(3);
+        q.Close();
+	DelayedEnqueuer de = new DelayedEnqueuer(q, 0, 4);
+        ExpectEof(new Thunk(de.EnqueueValue));
+        Assert.AreEqual(1, q.Dequeue());
+        Assert.AreEqual(2, q.DequeueNoWait(0));
+        bool r = q.Dequeue(1, out v);
+        Assert.IsTrue(r);
+        Assert.AreEqual(3, v);
+        ExpectEof(new Thunk(de.Dequeue));
+    }
+
+    [Test]
+    public void TestCloseWhenWaiting() {
+        SharedQueue q = new SharedQueue();
+	DelayedEnqueuer de = new DelayedEnqueuer(q, 0, null);
+        Thread t =
+	    new Thread(new ThreadStart(de.BackgroundEofExpectingDequeue));
+        t.Start();
+        Thread.Sleep(10);
+        q.Close();
+        t.Join();
+    }
 }

src/util/SharedQueue.cs

@@ -61,7 +61,7 @@
 
 namespace RabbitMQ.Util {
     ///<summary>A thread-safe shared queue implementation.</summary>
-    public class SharedQueue: IDisposable {
+    public class SharedQueue {
         ///<summary>The shared queue.</summary>
         ///<remarks>
         ///Subclasses must ensure appropriate locking discipline when
@@ -77,23 +77,17 @@ public class SharedQueue: IDisposable {
         public SharedQueue() {
         }
 
-        ///<summary>Close the queue. Causes all waiting threads to be
-        ///interrupted (with EndOfStreamException) and all further
-        ///Enqueue() and Dequeue() operations to throw
-        ///EndOfStreamException.</summary>
+        ///<summary>Close the queue. Causes all further Enqueue()
+        ///operations to throw EndOfStreamException, and all pending
+        ///or subsequent Dequeue() operations to throw an
+        ///EndOfStreamException once the queue is empty.</summary>
         public void Close() {
             lock (m_queue) {
                 m_isOpen = false;
                 Monitor.PulseAll(m_queue);
             }
         }
 
-        ///<summary>Implement IDisposable.Dispose. Delegates directly
-        ///to Close().</summary>
-        public void Dispose() {
-            Close();
-        }
-
         ///<summary>Call only when the lock on m_queue is held.</summary>
         /// <exception cref="EndOfStreamException" />
         private void EnsureIsOpen() {
@@ -121,12 +115,11 @@ public void Enqueue(object o) {
         ///<remarks>
         ///Callers of Dequeue() will block if no items are available
         ///until some other thread calls Enqueue() or the queue is
-        ///closed. If the queue is closed (by a call to Close()), this
-        ///method will throw EndOfStreamException.
+        ///closed. In the latter case this method will throw
+        ///EndOfStreamException.
         ///</remarks>
         public object Dequeue() {
             lock (m_queue) {
-                EnsureIsOpen();
                 while (m_queue.Count == 0) {
                     EnsureIsOpen();
                     Monitor.Wait(m_queue);
@@ -151,15 +144,15 @@ public object Dequeue() {
         /// whereas Dequeue() will.
         ///</para>
         ///<para>
-        /// If, at the time of call, the queue is in a closed state
-        /// (by a call to Close()), this method will throw
-        /// EndOfStreamException.
+        /// If at the time of call the queue is empty and in a
+        /// closed state (following a call to Close()), then this
+        /// method will throw EndOfStreamException.
         ///</para>
         ///</remarks>
         public object DequeueNoWait(object defaultValue) {
             lock (m_queue) {
-                EnsureIsOpen();
                 if (m_queue.Count == 0) {
+                    EnsureIsOpen();
                     return defaultValue;
                 } else {
                     return m_queue.Dequeue();
@@ -197,7 +190,8 @@ public object DequeueNoWait(object defaultValue) {
 	/// System.Threading.Monitor.Wait(object,int).
         ///</para>
         ///<para>
-        /// If, at any time during the call, the queue is in or
+        /// If no items are present and the queue is in a closed
+        /// state, or if at any time while waiting the queue
         /// transitions to a closed state (by a call to Close()), this
         /// method will throw EndOfStreamException.
         ///</para>
@@ -210,8 +204,8 @@ public bool Dequeue(int millisecondsTimeout, out object result) {
 
 	    DateTime startTime = DateTime.Now;
 	    lock (m_queue) {
-                EnsureIsOpen();
                 while (m_queue.Count == 0) {
+                    EnsureIsOpen();
 		    int elapsedTime = (int) ((DateTime.Now - startTime).TotalMilliseconds);
 		    int remainingTime = millisecondsTimeout - elapsedTime;
 		    if (remainingTime <= 0) {
@@ -220,7 +214,6 @@ public bool Dequeue(int millisecondsTimeout, out object result) {
 		    }
 
                     Monitor.Wait(m_queue, remainingTime);
-                    EnsureIsOpen();
                 }
 
                 result = m_queue.Dequeue();