Poller.remove socket disposed fix

src/NetMQ.Tests/NetMQPollerTest.cs

@@ -72,12 +72,12 @@ public void Monitoring()
                 int port = rep.BindRandomPort("tcp://127.0.0.1");
 
                 reqMonitor.Connected += (s, e) => connectedEvent.Set();
+                req.Connect("tcp://127.0.0.1:" + port);
 
                 reqMonitor.AttachToPoller(poller);
 
                 poller.RunAsync();
 
-                req.Connect("tcp://127.0.0.1:" + port);
                 req.SendFrame("a");
 
                 rep.SkipFrame();
@@ -347,6 +347,8 @@ public void RemoveSocket()
                     // identity
                     e.Socket.SkipFrame();
 
+                    //**Note: bad to assert from worker thread!  
+                    // If it fails, the test will crash, not report failure!
                     Assert.Equal("Hello", e.Socket.ReceiveFrameString(out bool more));
                     Assert.False(more);
 
@@ -400,6 +402,151 @@ public void RemoveSocket()
             }
         }
 
+        [Fact]
+        public async Task RemoveAndDisposeSocketAsync()
+        {
+            //set up poller, start it
+            var patient = new NetMQPoller();
+            patient.RunAsync();
+
+            Assert.True(patient.IsRunning);
+
+            //create a pub-sub pair
+            var port = 55667;
+            var conn = $"tcp://127.0.0.1:{port}";
+
+            var pub = new PublisherSocket();
+            pub.Bind(conn);
+
+            var sub = new SubscriberSocket();
+            sub.Connect(conn);
+            sub.SubscribeToAnyTopic();
+
+            //handle callbacks from poller thread
+            sub.ReceiveReady += (s, e) =>
+            {
+                var msg = e.Socket.ReceiveFrameString();
+
+                Debug.WriteLine($"sub has data: {msg}");
+            };
+
+            //add the subscriber socket to poller
+            patient.Add(sub);
+
+            //set up pub on separate thread
+            var canceller = new CancellationTokenSource();
+
+            var pubAction = new Action(async () =>
+            {
+                var token = canceller.Token;
+
+                uint i = 0;
+
+                while (!token.IsCancellationRequested)
+                {
+                    pub.SendFrame($"Hello-{++i}");
+
+                    // send ~ 5Hz
+                    await Task.Delay(200);
+                }
+            });
+
+            Task.Run(pubAction);
+
+            //allow a little time to run
+            await Task.Delay(2000);
+
+            //now try to remove the sub from poller
+            await patient.RemoveAndDisposeAsync(sub);
+
+            Assert.True(sub.IsDisposed);
+
+            //allow for poller to continue running
+            await Task.Delay(2000);
+
+            patient.Stop();
+            Assert.False(patient.IsRunning);
+
+            canceller.Cancel();
+
+            pub?.Dispose();
+            patient?.Dispose();
+        }
+
+        [Fact]
+        public async Task DisposeSocketAfterAsyncRemoval()
+        {
+            //set up poller, start it
+            var patient = new NetMQPoller();
+            patient.RunAsync();
+
+            Assert.True(patient.IsRunning);
+
+            //create a pub-sub pair
+            var port = 55667;
+            var conn = $"tcp://127.0.0.1:{port}";
+
+            var pub = new PublisherSocket();
+            pub.Bind(conn);
+
+            var sub = new SubscriberSocket();
+            sub.Connect(conn);
+            sub.SubscribeToAnyTopic();
+
+            //handle callbacks from poller thread
+            sub.ReceiveReady += (s, e) =>
+            {
+                var msg = e.Socket.ReceiveFrameString();
+
+                Debug.WriteLine($"sub has data: {msg}");
+            };
+
+            //add the subscriber socket to poller
+            patient.Add(sub);
+
+            //set up pub on separate thread
+            var canceller = new CancellationTokenSource();
+
+            var pubAction = new Action(async () =>
+            {
+                var token = canceller.Token;
+
+                uint i = 0;
+
+                while(!token.IsCancellationRequested)
+                {
+                    pub.SendFrame($"Hello-{++i}");
+
+                    // send ~ 5Hz
+                    await Task.Delay(200);
+                }
+            });
+
+            var pubThread = Task.Run(pubAction);
+
+            //allow a little time to run
+            await Task.Delay(2000);
+
+            //now try to remove the sub from poller
+            await patient.RemoveAsync(sub);
+
+            // dispose the sub (this will cause exception on poller's worker-thread) and it can't be caught!
+            sub.Dispose();
+            sub = null;
+
+            //allow for poller to continue running
+            await Task.Delay(2000);
+
+            patient.Stop();
+            Assert.False(patient.IsRunning);
+
+            canceller.Cancel();
+
+            pub?.Dispose();
+            patient?.Dispose();
+            canceller?.Dispose();
+        }
+
         [Fact]
         public void AddThrowsIfSocketAlreadyDisposed()
         {
@@ -453,6 +600,7 @@ public void DisposeThrowsIfSocketAlreadyDisposed()
             // Dispose the socket.
             // It is incorrect to have a disposed socket in a poller.
             // Disposed sockets can throw into the poller's thread.
+
             socket.Dispose();
 
             // Dispose throws if a polled socket is disposed

src/NetMQ/ISocketPollableCollection.cs

@@ -10,7 +10,7 @@ namespace NetMQ
     /// <remarks>
     /// This interface provides an abstraction over the legacy Poller and newer <see cref="NetMQPoller"/> classes for use in <see cref="NetMQMonitor"/>.
     /// </remarks>
-    [Obsolete("Use INetMQPoller instead")]
+    [Obsolete("Use INetMQPoller instead. This should be made internal, to avoid major re-work of NetMQMonitor, but prevent accidental mis-use from applications")]
     public interface ISocketPollableCollection
     {
         /// <summary>

src/NetMQ/ISocketPollableCollectionAsync.cs

@@ -0,0 +1,18 @@
+using System;
+using System.Threading.Tasks;
+using JetBrains.Annotations;
+
+namespace NetMQ
+{
+    /// <summary>
+    ///
+    /// </summary>
+    /// <remarks>
+    /// This interface provides an abstraction over the legacy Poller and newer <see cref="NetMQPoller"/> classes for use in <see cref="NetMQMonitor"/> and avoids thread sync issues removing sockets.
+    /// </remarks>
+    public interface ISocketPollableCollectionAsync
+    {
+        Task RemoveAsync([NotNull] ISocketPollable socket);
+        Task RemoveAndDisposeAsync<T>(T socket) where T : ISocketPollable, IDisposable;
+    }
+}

src/NetMQ/NetMQ.csproj

@@ -57,4 +57,43 @@
     <Reference Include="System" />
     <Reference Include="Microsoft.CSharp" />
   </ItemGroup>
+
+<!--
+  <ItemGroup Condition=" '$(TargetFramework)' == 'monoandroid60' ">
+    <Reference Include="mscorlib" />
+    <Reference Include="System" />
+    <Reference Include="System.Core" />
+    <Reference Include="System.ServiceModel" />
+  </ItemGroup>
+
+  <ItemGroup Condition=" '$(TargetFramework)' == 'xamarinios10' ">
+    <Reference Include="mscorlib" />
+    <Reference Include="System" />
+    <Reference Include="System.Core" />
+    <Reference Include="System.ServiceModel" />
+  </ItemGroup>
+-->
+
+  <ItemGroup>
+    <PackageReference Include="SourceLink.Create.CommandLine" Version="2.7.4" PrivateAssets="All" />
+  </ItemGroup>
+
+  <ItemGroup Condition="'$(TargetFramework)' == 'net47'">
+    <PackageReference Include="Microsoft.Bcl.Async">
+      <Version>1.0.168</Version>
+    </PackageReference>
+  </ItemGroup>
+
+  <ItemGroup Condition="'$(TargetFramework)' == 'net40'">
+    <PackageReference Include="Microsoft.Bcl.Async">
+      <Version>1.0.168</Version>
+    </PackageReference>
+  </ItemGroup>
+
+  <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'">
+    <PackageReference Include="Microsoft.Bcl.Async">
+      <Version>1.0.168</Version>
+    </PackageReference>
+  </ItemGroup>
+
 </Project>