Merge pull request #766 from AbirAtur/queue_capacity_bugfix

fix for #759

Author: somdoron

src/NetMQ.Tests/NetMQQueueTests.cs

@@ -1,6 +1,9 @@
 #if !NET35
 using System;
+using System.Net.Sockets;
 using System.Threading;
+using System.Threading.Tasks;
+using NetMQ.Sockets;
 using Xunit;
 
 namespace NetMQ.Tests
@@ -20,6 +23,27 @@ public void EnqueueDequeue()
             }
         }
 
+        [Fact]
+        public void EnqueueShouldNotBlockWhenCapacityIsZero()
+        {
+            using (var mockSocket = new PairSocket())
+            using (var queue = new NetMQQueue<int>())
+            {
+                int socketWatermarkCapacity = mockSocket.Options.SendHighWatermark + mockSocket.Options.ReceiveHighWatermark;
+
+                Task task = Task.Run(() =>
+                {
+                    for (int i = 0; i < socketWatermarkCapacity + 100; i++)
+                    {
+                        queue.Enqueue(i);
+                    }
+                });
+
+                bool completed = task.Wait(TimeSpan.FromSeconds(1));
+                Assert.True(completed, "Enqueue task should have completed " + socketWatermarkCapacity + " enqueue within 1 second");
+            }
+        }
+
         [Fact]
         public void TryDequeue()
         {

src/NetMQ/NetMQQueue.cs

@@ -35,9 +35,10 @@ public NetMQQueue(int capacity = 0)
                 throw new ArgumentOutOfRangeException(nameof(capacity));
 
             m_queue = new ConcurrentQueue<T>();
-            PairSocket.CreateSocketPair(out m_writer, out m_reader);
-
-            m_writer.Options.SendHighWatermark = m_reader.Options.ReceiveHighWatermark = capacity / 2;
+            PairSocket.CreateSocketPair(out m_writer,
+                                        out m_reader,
+                                        writer => writer.Options.SendHighWatermark = capacity / 2,
+                                        reader => reader.Options.ReceiveHighWatermark = capacity / 2);
 
             m_eventDelegator = new EventDelegator<NetMQQueueEventArgs<T>>(
                 () => m_reader.ReceiveReady += OnReceiveReady,

src/NetMQ/Sockets/PairSocket.cs

@@ -1,4 +1,5 @@
-using System.Threading;
+using System;
+using System.Threading;
 using NetMQ.Core;
 
 namespace NetMQ.Sockets
@@ -44,5 +45,25 @@ public static void CreateSocketPair(out PairSocket socket1, out PairSocket socke
             socket2 = new PairSocket();
             socket2.Connect(address);
         }
+
+        /// <summary>
+        /// Create and return an inproc pipe where socket1 is bound and socket2 is connected.
+        /// </summary>
+        /// <param name="socket1">the Bind socket</param>
+        /// <param name="socket2">the Connect socket</param>
+        /// <param name="initSocket1">Method to initialize socket1 before connection</param>
+        /// <param name="initSocket2">Method to initialize socket2 before connection</param>
+        public static void CreateSocketPair(out PairSocket socket1, out PairSocket socket2, Action<PairSocket> initSocket1, Action<PairSocket> initSocket2)
+        {
+            string address = $"inproc://NetMQSocketPair#{Interlocked.Increment(ref s_sequence)}";
+
+            socket1 = new PairSocket();
+            initSocket1(socket1);
+            socket1.Bind(address);
+
+            socket2 = new PairSocket();
+            initSocket2(socket2);
+            socket2.Connect(address);
+        }
     }
 }