merging bug 21468 into default

Author: Matthew Sackman

projects/client/RabbitMQ.Client/src/util/SharedQueue.cs

@@ -61,7 +61,7 @@
 
 namespace RabbitMQ.Util {
     ///<summary>A thread-safe shared queue implementation.</summary>
-    public class SharedQueue {
+    public class SharedQueue : IEnumerable {
         ///<summary>The shared queue.</summary>
         ///<remarks>
         ///Subclasses must ensure appropriate locking discipline when
@@ -220,5 +220,55 @@ public bool Dequeue(int millisecondsTimeout, out object result) {
 		return true;
             }
 	}
+
+        ///<summary>Implementation of the IEnumerable interface, for
+        ///permitting SharedQueue to be used in foreach
+        ///loops.</summary>
+        IEnumerator IEnumerable.GetEnumerator() {
+            return new SharedQueueEnumerator(this);
+        }
+
     }
+
+    ///<summary>Implementation of the IEnumerator interface, for
+    ///permitting SharedQueue to be used in foreach loops.</summary>
+    public class SharedQueueEnumerator : IEnumerator {
+
+        protected SharedQueue m_queue;
+        protected object m_current;
+
+        ///<summary>Construct an enumerator for the given
+        ///SharedQueue.</summary>
+        public SharedQueueEnumerator(SharedQueue queue) {
+            m_queue = queue;
+        }
+
+        object IEnumerator.Current {
+            get {
+                if (m_current == null) {
+                    throw new InvalidOperationException();
+                }
+                return m_current;
+            }
+        }
+
+        bool IEnumerator.MoveNext() {
+            try {
+                m_current = m_queue.Dequeue();
+                return true;
+            } catch (EndOfStreamException) {
+                m_current = null;
+                return false;
+            }
+        }
+
+        ///<summary>Reset()ting a SharedQueue doesn't make sense, so
+        ///this method always throws
+        ///InvalidOperationException.</summary>
+        void IEnumerator.Reset() {
+            throw new InvalidOperationException("SharedQueue.Reset() does not make sense");
+        }
+
+    }
+
 }

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

@@ -72,6 +72,11 @@ public class TestSharedQueue
 
         public delegate void Thunk();
 
+        //wrapper to work around C#'s lack of local volatiles
+        public class VolatileInt {
+            public volatile int v = 0;
+        }
+
         public class DelayedEnqueuer
         {
             public SharedQueue m_q;
@@ -342,6 +347,30 @@ public void TestCloseWhenWaiting()
             q.Close();
             t.Join();
         }
+
+        [Test]
+        public void TestEnumerator()
+        {
+            SharedQueue q = new SharedQueue();
+            VolatileInt c1 = new VolatileInt();
+            VolatileInt c2 = new VolatileInt();
+            Thread t1 = new Thread(delegate() {
+                    foreach (int v in q) c1.v+=v;
+                });
+            Thread t2 = new Thread(delegate() {
+                    foreach (int v in q) c2.v+=v;
+                });
+            t1.Start();
+            t2.Start();
+            q.Enqueue(1);
+            q.Enqueue(2);
+            q.Enqueue(3);
+            q.Close();
+            t1.Join();
+            t2.Join();
+            Assert.AreEqual(6, c1.v + c2.v);
+        }
+
     }
 
-}
+}