permit Dequeue()ing after Close() until queue is empty

EndOfStreamException is now only thrown by Dequeue*() methods when the
queue is/gets closed *and* is empty.

Author: Matthias Radestock

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/util/SharedQueue.cs

@@ -77,10 +77,10 @@ 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;
@@ -121,12 +121,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 +150,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 +196,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 is
         /// transitions to a closed state (by a call to Close()), this
         /// method will throw EndOfStreamException.
         ///</para>
@@ -210,8 +210,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 +220,6 @@ public bool Dequeue(int millisecondsTimeout, out object result) {
 		    }
 
                     Monitor.Wait(m_queue, remainingTime);
-                    EnsureIsOpen();
                 }
 
                 result = m_queue.Dequeue();