Merge pull request #7 from somdoron/master

cherry picks fixes from NetMQ4

src/NetMQ.Tests/CleanupTests.cs

@@ -0,0 +1,71 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using NetMQ.Sockets;
+using NUnit.Framework;
+
+namespace NetMQ.Tests
+{
+    [TestFixture]
+    public class CleanupTests
+    {
+        [Test]
+        public void Block()
+        {
+            const int count = 1000;
+
+            NetMQConfig.Linger = TimeSpan.FromSeconds(0.5);           
+
+            using (var client = new DealerSocket(">tcp://localhost:5557"))
+            {                
+                // Sending a lot of messages
+                client.Options.SendHighWatermark = count;
+                for (int i = 0; i < count; i++)
+                {
+                    client.SendFrame("Hello");
+                }                
+            }
+
+            Stopwatch stopwatch = Stopwatch.StartNew();
+            NetMQConfig.Cleanup();
+            stopwatch.Stop();
+
+            Assert.Greater(stopwatch.ElapsedMilliseconds, 500);
+        }
+
+        [Test]
+        public void NoBlock()
+        {
+            const int count = 1000;
+
+            NetMQConfig.Linger = TimeSpan.FromSeconds(0.5);
+
+            using (var client = new DealerSocket(">tcp://localhost:5557"))
+            {
+                // Sending a lot of messages
+                client.Options.SendHighWatermark = count;
+                for (int i = 0; i < count; i++)
+                {
+                    client.SendFrame("Hello");
+                }
+            }
+
+            Stopwatch stopwatch = Stopwatch.StartNew();
+            NetMQConfig.Cleanup(false);
+            stopwatch.Stop();
+
+            Assert.Less(stopwatch.ElapsedMilliseconds, 500);
+        }
+
+        [Test]
+        public void NoBlockNoDispose()
+        {
+            var client = new DealerSocket(">tcp://localhost:5557");
+            NetMQConfig.Cleanup(false);
+        }
+    }
+}

src/NetMQ.Tests/NetMQ.Tests.csproj

@@ -72,6 +72,7 @@
     <Compile Include="ActorTests.cs" />
     <Compile Include="BeaconTests.cs" />
     <Compile Include="ByteArraySegmentTests.cs" />
+    <Compile Include="CleanupTests.cs" />
     <Compile Include="EventDelegatorTests.cs" />
     <Compile Include="ExceptionTests.cs" />
     <Compile Include="InProcActors\AccountJSON\Account.cs" />
@@ -91,6 +92,7 @@
     <Compile Include="ReceivingSocketExtensionsTests.cs" />
     <Compile Include="RequestWithRetryTests.cs" />
     <Compile Include="RouterTests.cs" />
+    <Compile Include="Setup.cs" />
     <Compile Include="XPubSubTests.cs" />
     <Compile Include="Devices\DeviceTestBase.cs" />
     <Compile Include="Devices\ForwarderDeviceTests.cs" />

src/NetMQ.Tests/NetMQ3.5.Tests.csproj

@@ -73,6 +73,7 @@
     <Compile Include="ActorTests.cs" />
     <Compile Include="BeaconTests.cs" />
     <Compile Include="ByteArraySegmentTests.cs" />
+    <Compile Include="CleanupTests.cs" />
     <Compile Include="Core\YQueueTests.cs" />
     <Compile Include="EventDelegatorTests.cs" />
     <Compile Include="ExceptionTests.cs" />
@@ -84,6 +85,7 @@
     <Compile Include="NetMQPollerTest.cs" />
     <Compile Include="ReceivingSocketExtensionsTests.cs" />
     <Compile Include="RouterTests.cs" />
+    <Compile Include="Setup.cs" />
     <Compile Include="SocketOptionsTests.cs" />
     <Compile Include="XPubSubTests.cs" />
     <Compile Include="MessageTests.cs" />

src/NetMQ.Tests/NetMQMonitorTests.cs

@@ -8,15 +8,15 @@
 
 namespace NetMQ.Tests
 {
-    [TestFixture]
+    [TestFixture(Category = "Monitor")]
     public class NetMQMonitorTests
     {
         [Test]
         public void Monitoring()
         {            
             using (var rep = new ResponseSocket())
             using (var req = new RequestSocket())
-            using (var monitor = new NetMQMonitor(rep, "inproc://rep.inproc", SocketEvents.Accepted | SocketEvents.Listening))
+            using (var monitor = new NetMQMonitor(rep, $"inproc://rep.inproc", SocketEvents.Accepted | SocketEvents.Listening))
             {
                 var listening = false;
                 var accepted = false;
@@ -28,6 +28,8 @@ public void Monitoring()
 
                 var monitorTask = Task.Factory.StartNew(monitor.Start);
 
+                Thread.Sleep(10);
+
                 var port = rep.BindRandomPort("tcp://127.0.0.1");
 
                 req.Connect("tcp://127.0.0.1:" + port);
@@ -47,8 +49,8 @@ public void Monitoring()
 
                 Thread.Sleep(200);
 
-                Assert.IsTrue(monitorTask.IsCompleted);
-            }
+                Assert.IsTrue(monitorTask.IsCompleted);                
+            }            
         }
 
 #if !NET35
@@ -102,6 +104,7 @@ public void ErrorCodeTest()
                 monitor.Timeout = TimeSpan.FromMilliseconds(100);
 
                 var monitorTask = Task.Factory.StartNew(monitor.Start);
+                Thread.Sleep(10);
 
                 var port = rep.BindRandomPort("tcp://127.0.0.1");
 

src/NetMQ.Tests/Setup.cs

@@ -0,0 +1,18 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using NUnit.Framework;
+
+namespace NetMQ.Tests
+{
+    [SetUpFixture]
+    public class Setup
+    {
+        [TearDown]
+        public void TearDown()
+        {
+            NetMQConfig.Cleanup(false);
+        }
+    }
+}

src/NetMQ/Core/Command.cs

@@ -52,7 +52,7 @@ public Command([CanBeNull] ZObject destination, CommandType type, [CanBeNull] ob
         /// Get the argument to this command.
         /// </summary>
         [CanBeNull]
-        public object Arg { get; private set; }
+        public object Arg { get; private set; }        
 
         /// <summary>
         /// Override of ToString, which returns a string in the form [ command-type, destination ].

src/NetMQ/Core/CommandType.cs

@@ -101,6 +101,11 @@ internal enum CommandType
         /// Sent by reaper thread to the term thread when all the sockets
         /// have successfully been deallocated.
         /// </summary>
-        Done
+        Done,
+
+        /// <summary>
+        /// Send to reaper to stop the reaper immediatly
+        /// </summary>
+        ForceStop
     }
 }

src/NetMQ/Core/Ctx.cs

@@ -36,8 +36,8 @@ namespace NetMQ.Core
     /// <remarks>Internal analog of the public <see cref="NetMQContext"/> class.</remarks>
     internal sealed class Ctx
     {
-        private const int DefaultIOThreads = 1;
-        private const int DefaultMaxSockets = 1024;
+        internal const int DefaultIOThreads = 1;
+        internal const int DefaultMaxSockets = 1024;
 
         #region Nested class: Endpoint
 
@@ -217,8 +217,12 @@ public void Terminate()
                         foreach (var socket in m_sockets)
                             socket.Stop();
 
-                        if (m_sockets.Count == 0)
-                            m_reaper.Stop();
+                        if (!Block)
+                        {
+                            m_reaper.ForceStop();
+                        }
+                        else if (m_sockets.Count == 0)
+                            m_reaper.Stop();                        
                     }
                     finally
                     {
@@ -234,8 +238,7 @@ public void Terminate()
 
                     Debug.Assert(found);
                     Debug.Assert(command.CommandType == CommandType.Done);
-                    Monitor.Enter(m_slotSync);
-                    Debug.Assert(m_sockets.Count == 0);
+                    Monitor.Enter(m_slotSync);                
                 }
                 else
                     Monitor.Enter(m_slotSync);

src/NetMQ/Core/Reaper.cs

@@ -114,6 +114,11 @@ public void Stop()
                 SendStop();
         }
 
+        public void ForceStop()
+        {
+            SendForceStop();
+        }
+
         /// <summary>
         /// Handle input-ready events, by receiving and processing any commands
         /// that are waiting in the mailbox.
@@ -168,6 +173,14 @@ protected override void ProcessStop()
             }
         }
 
+        protected override void ProcessForceStop()
+        {
+            m_terminating = true;
+            SendDone();
+            m_poller.RemoveHandle(m_mailboxHandle);
+            m_poller.Stop();
+        }
+
         /// <summary>
         /// Add the given socket to the list to be reaped (terminated).
         /// </summary>
@@ -196,6 +209,6 @@ protected override void ProcessReaped()
                 m_poller.RemoveHandle(m_mailboxHandle);
                 m_poller.Stop();
             }
-        }
+        }        
     }
 }

src/NetMQ/Core/Utils/Poller.cs

@@ -135,8 +135,7 @@ public void Destroy()
             if (!m_stopped)
             {
                 try
-                {
-                    m_stopping = true;
+                {                    
                     m_workerThread.Join();
                 }
                 catch (Exception)

src/NetMQ/Core/ZObject.cs

@@ -125,6 +125,11 @@ protected void SendStop()
             m_ctx.SendCommand(m_threadId, new Command(this, CommandType.Stop));
         }
 
+        protected void SendForceStop()
+        {
+            m_ctx.SendCommand(m_threadId, new Command(this, CommandType.ForceStop));
+        }
+
         /// <summary>
         /// Send the Plug command, incrementing the destinations sequence-number if incSeqnum is true.
         /// </summary>
@@ -322,6 +327,10 @@ public void ProcessCommand([NotNull] Command cmd)
                     ProcessReaped();
                     break;
 
+                case CommandType.ForceStop:
+                    ProcessForceStop();
+                    break;
+
                 default:
                     throw new ArgumentException();
             }
@@ -333,6 +342,12 @@ protected virtual void ProcessStop()
             throw new NotSupportedException();
         }
 
+        /// <exception cref="NotSupportedException">Not supported on the ZObject class.</exception>
+        protected virtual void ProcessForceStop()
+        {
+            throw new NotSupportedException();
+        }
+
         /// <exception cref="NotSupportedException">Not supported on the ZObject class.</exception>
         protected virtual void ProcessPlug()
         {