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
@@ -562,6 +710,61 @@ public void RemoveTimer()
  }
  }
 
+ [Fact]
+ public async void RemoveTimer_Async()
+ {
+ using (var router = new RouterSocket())
+ using (var dealer = new DealerSocket())
+ using (var poller = new NetMQPoller { router })
+ {
+ int port = router.BindRandomPort("tcp://127.0.0.1");
+
+ dealer.Connect("tcp://127.0.0.1:" + port);
+
+ bool timerTriggered = false;
+
+ var timer = new NetMQTimer(TimeSpan.FromMilliseconds(500));
+ timer.Elapsed += (s, a) => { timerTriggered = true; };
+
+ // The timer will fire after 100ms
+ poller.Add(timer);
+
+ bool messageArrived = false;
+
+ router.ReceiveReady += (s, e) =>
+ {
+ router.SkipFrame();
+ router.SkipFrame();
+ messageArrived = true;
+ //// Remove timer
+ //poller.Remove(timer);
+ };
+
+ poller.RunAsync();
+
+ Thread.Sleep(20);
+
+ dealer.SendFrame("hello");
+
+ Thread.Sleep(300);
+
+ try
+ {
+ await poller.RemoveAsync(timer);
+ }
+ catch (Exception ex)
+ {
+ //ensure no exceptions thrown
+ Assert.Null(ex);
+ }
+
+ poller.Stop();
+
+ Assert.True(messageArrived);
+ Assert.False(timerTriggered);
+ }
+ }
+
  [Fact]
  public void RunMultipleTimes()
  {
@@ -859,6 +1062,60 @@ public void NativeSocket()
  }
  }
 
+ [Fact]
+ public async void NativeSocket_RemoveAsync()
+ {
+ using (var streamServer = new StreamSocket())
+ using (var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
+ {
+ int port = streamServer.BindRandomPort("tcp://*");
+
+ socket.Connect(new IPEndPoint(IPAddress.Parse("127.0.0.1"), port));
+
+ var buffer = new byte[] { 1 };
+ socket.Send(buffer);
+
+ byte[] identity = streamServer.ReceiveFrameBytes();
+ byte[] message = streamServer.ReceiveFrameBytes();
+
+ Assert.Equal(buffer[0], message[0]);
+
+ var socketSignal = new ManualResetEvent(false);
+
+ using (var poller = new NetMQPoller())
+ {
+ poller.Add(socket, s =>
+ {
+ socket.Receive(buffer);
+
+ socketSignal.Set();
+ });
+
+ poller.RunAsync();
+
+ // no message is waiting for the socket so it should fail
+ Assert.False(socketSignal.WaitOne(100));
+
+ // sending a message back to the socket
+ streamServer.SendMoreFrame(identity).SendFrame("a");
+
+ Assert.True(socketSignal.WaitOne(100));
+
+ socketSignal.Reset();
+
+ await poller.RemoveAsync(socket);
+
+ // sending a message back to the socket
+ streamServer.SendMoreFrame(identity).SendFrame("a");
+
+ // we remove the native socket so it should fail
+ Assert.False(socketSignal.WaitOne(100));
+
+ poller.Stop();
+ }
+ }
+ }
+
  #endregion
 
  #region TaskScheduler tests

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>