Merge branch 'zeromq:master' into master

Author: mmillerbe

README.md

@@ -25,7 +25,7 @@ This repository is for version 4, for version 3 go to: https://github.com/NetMQ/
 
 ## Using / Documentation
 
-Before using NetMQ, make sure to read the [ZeroMQ Guide](http://zguide.zeromq.org/page:all).
+Before using NetMQ, make sure to read the [ZeroMQ Guide](http://zguide.zeromq.org/).
 
 The NetMQ documentation can be found at [netmq.readthedocs.org](http://netmq.readthedocs.org/en/latest/). Thanks to [Sacha Barber](http://www.codeproject.com/Members/Sacha-Barber) who agreed to do the documentation.
 

src/NetMQ.Tests/RouterTests.cs

@@ -130,5 +130,26 @@ public void Handover()
                 }
             }
         }
+
+        [Fact]
+        public void RoutingKeys() 
+        {
+            using var router = new RouterSocket("inproc://routing-keys");
+            using var dealer = new DealerSocket("inproc://routing-keys");
+
+            dealer.SendRoutingKeys(new RoutingKey(1)).SendFrame("Hello");
+            
+            var keys = router.ReceiveRoutingKeys();
+            var message = router.ReceiveFrameString();
+
+            Assert.Equal("Hello", message);
+
+            router.SendRoutingKeys(keys).SendFrame("World");
+
+            dealer.ReceiveRoutingKeys();
+            var reply = dealer.ReceiveFrameString();
+
+            Assert.Equal("World", reply);
+        }
     }
 }

src/NetMQ/AsyncReceiveExtensions.cs

@@ -20,7 +20,7 @@ public static class AsyncReceiveExtensions
         static Task<bool> s_trueTask = Task.FromResult(true);
         static Task<bool> s_falseTask = Task.FromResult(false);
 
-        #region Receiving frames as a multipart message
+#region Receiving frames as a multipart message
 
         /// <summary>
         /// Receive a single frame from <paramref name="socket"/>, asynchronously.
@@ -52,9 +52,9 @@ public static async Task<NetMQMessage> ReceiveMultipartMessageAsync(
             return message;
         }
 
-        #endregion
+#endregion
 
-        #region Receiving a frame as a byte array
+#region Receiving a frame as a byte array
 
         /// <summary>
         /// Receive a single frame from <paramref name="socket"/>, asynchronously.
@@ -85,7 +85,12 @@ public static async Task<NetMQMessage> ReceiveMultipartMessageAsync(
             }
 
             TaskCompletionSource<(byte[], bool)> source = new TaskCompletionSource<(byte[], bool)>();
-            cancellationToken.Register(() => source.SetCanceled());
+
+            CancellationTokenRegistration? registration = null;
+            if (cancellationToken.CanBeCanceled) 
+            {
+                registration = cancellationToken.Register(PropagateCancel);
+            }
 
             void Listener(object sender, NetMQSocketEventArgs args)
             {
@@ -96,18 +101,26 @@ void Listener(object sender, NetMQSocketEventArgs args)
                     msg.Close();
 
                     socket.ReceiveReady -=  Listener;
-                    source.SetResult((data, more));
+                    registration?.Dispose();
+                    source.TrySetResult((data, more));
                 }
             }
+            
+            void PropagateCancel()
+            {
+                socket.ReceiveReady -= Listener;
+                registration?.Dispose();
+                source.TrySetCanceled();
+            }
 
             socket.ReceiveReady += Listener;
 
             return source.Task;
         }
 
-        #endregion
+#endregion
 
-        #region Receiving a frame as a string
+#region Receiving a frame as a string
 
         /// <summary>
         /// Receive a single frame from <paramref name="socket"/>, asynchronously, and decode as a string using <see cref="SendReceiveConstants.DefaultEncoding"/>.
@@ -155,7 +168,12 @@ void Listener(object sender, NetMQSocketEventArgs args)
             }
 
             TaskCompletionSource<(string, bool)> source = new TaskCompletionSource<(string,bool)>();
-            cancellationToken.Register(() => source.SetCanceled());
+
+            CancellationTokenRegistration? registration = null;
+            if (cancellationToken.CanBeCanceled) 
+            {
+                registration = cancellationToken.Register(PropagateCancel);
+            }
 
             void Listener(object sender, NetMQSocketEventArgs args)
             {
@@ -165,28 +183,40 @@ void Listener(object sender, NetMQSocketEventArgs args)
                         ? msg.GetString(encoding)
                         : string.Empty;
                     bool more = msg.HasMore;
-
                     msg.Close();
+
                     socket.ReceiveReady -=  Listener;
-                    source.SetResult((str, more));
+                    registration?.Dispose();
+                    source.TrySetResult((str, more));
                 }
             }
 
+            void PropagateCancel()
+            {
+                socket.ReceiveReady -= Listener;
+                registration?.Dispose();
+                source.TrySetCanceled();
+            }
+
             socket.ReceiveReady += Listener;
 
             return source.Task;
         }
 
-        #endregion
+#endregion
 
-        #region Skipping a message
+#region Skipping a message
 
         /// <summary>
         /// Receive a single frame from <paramref name="socket"/>, asynchronously, then ignore its content.
         /// </summary>
         /// <param name="socket">The socket to receive from.</param>
+        /// <param name="cancellationToken">The token used to propagate notification that this operation should be canceled.</param>
         /// <returns>Boolean indicate if another frame of the same message follows</returns>
-        public static Task<bool> SkipFrameAsync(this NetMQSocket socket)
+        public static Task<bool> SkipFrameAsync(
+            this NetMQSocket socket, 
+            CancellationToken cancellationToken = default(CancellationToken)
+        )
         {
             if (NetMQRuntime.Current == null)
                 throw new InvalidOperationException("NetMQRuntime must be created before calling async functions");
@@ -206,26 +236,41 @@ public static Task<bool> SkipFrameAsync(this NetMQSocket socket)
 
             TaskCompletionSource<bool> source = new TaskCompletionSource<bool>();
 
+            CancellationTokenRegistration? registration = null;
+            if (cancellationToken.CanBeCanceled) 
+            {
+                registration = cancellationToken.Register(PropagateCancel);
+            }
+
             void Listener(object sender, NetMQSocketEventArgs args)
             {
                 if (socket.TryReceive(ref msg, TimeSpan.Zero))
                 {
                     bool more = msg.HasMore;
                     msg.Close();
+
                     socket.ReceiveReady -=  Listener;
-                    source.SetResult(more);
+                    registration?.Dispose();
+                    source.TrySetResult(more);
                 }
             }
 
+            void PropagateCancel()
+            {
+                socket.ReceiveReady -= Listener;
+                registration?.Dispose();
+                source.TrySetCanceled();
+            }
+
             socket.ReceiveReady += Listener;
 
             return source.Task;
         }
 
 
-        #endregion
+#endregion
 
-        #region Skipping all frames of a multipart message
+#region Skipping all frames of a multipart message
 
         /// <summary>
         /// Receive all frames of the next message from <paramref name="socket"/>, asynchronously, then ignore their contents.
@@ -242,9 +287,9 @@ public static async Task SkipMultipartMessageAsync(this NetMQSocket socket)
         }
 
 
-        #endregion
+#endregion
 
-        #region Receiving a routing key
+#region Receiving a routing key
 
         /// <summary>
         /// Receive a routing-key from <paramref name="socket"/>, blocking until one arrives.
@@ -259,7 +304,7 @@ public static async Task SkipMultipartMessageAsync(this NetMQSocket socket)
             return (new RoutingKey(bytes), more);
         }
 
-        #endregion
+#endregion
     }
 }
 

src/NetMQ/Msg.cs

@@ -265,6 +265,18 @@ public Span<byte> Slice()
 
             return new Span<byte>(m_data, m_offset, Size);  
         }
+        
+        /// <summary>
+        /// Return the internal buffer as Memory
+        /// </summary>
+        /// <returns>The memory</returns>
+        public Memory<byte> SliceAsMemory()
+        {
+            if (m_data == null)
+                return Memory<byte>.Empty;
+            
+            return new Memory<byte>(m_data, m_offset, Size);  
+        }
 
         /// <summary>
         /// Returns a slice of the internal buffer.

src/NetMQ/NetMQ-unix.csproj

@@ -49,17 +49,9 @@
     <PackageReference Include="System.ServiceModel.Primitives" Version="4.4.0" />
     <PackageReference Include="System.ValueTuple" Version="4.5.0" />
   </ItemGroup>
-  
-  <ItemGroup Condition=" '$(TargetFramework)' == 'net45' ">
-    <PackageReference Include="Microsoft.NETFramework.ReferenceAssemblies.net45" Version="1.0.0" PrivateAssets="All" />
-    <PackageReference Include="System.Threading.Tasks.Extensions" Version="4.5.4" />
-    <Reference Include="System.ServiceModel" />
-    <Reference Include="System" />
-    <Reference Include="Microsoft.CSharp" />
-  </ItemGroup>
 
-  <ItemGroup Condition=" '$(TargetFramework)' == 'net47' ">
-    <PackageReference Include="Microsoft.NETFramework.ReferenceAssemblies.net47" Version="1.0.0" PrivateAssets="All" />
+  <ItemGroup Condition=" '$(TargetFrameworkIdentifier)' == '.NETFramework' ">
+    <PackageReference Include="Microsoft.NETFramework.ReferenceAssemblies" Version="1.0.0" PrivateAssets="All" />
     <PackageReference Include="System.Threading.Tasks.Extensions" Version="4.5.4" />
     <Reference Include="System.ServiceModel" />
     <Reference Include="System" />

src/NetMQ/NetMQ.csproj

@@ -199,7 +199,8 @@ private void OnUdpReady(Socket socket)
             {
                 Assumes.NotNull(m_pipe);
 
-                var frame = ReceiveUdpFrame(out string peerName);
+                if (!TryReceiveUdpFrame(out NetMQFrame frame, out string peerName))
+		    return;
 
                 // If filter is set, check that beacon matches it
                 var isValid = frame.MessageSize >= m_filter?.MessageSize && Compare(frame, m_filter, m_filter.MessageSize);
@@ -269,28 +270,37 @@ private void SendUdpFrame(NetMQFrame frame)
                 {
                     m_udpSocket.SendTo(frame.Buffer, 0, frame.MessageSize, SocketFlags.None, m_broadcastAddress);
                 }
-                catch (SocketException ex)
+                catch (SocketException ex) when (ex.SocketErrorCode == SocketError.AddressNotAvailable)
                 {
-                    if (ex.SocketErrorCode != SocketError.AddressNotAvailable) { throw; }
-
                     // Initiate Creation of new Udp here to solve issue related to 'sudden' network change.
                     // On windows (7 OR 10) incorrect/previous ip address might still exist instead of new Ip
                     // due to network change which causes crash (if no try/catch and keep trying to send to incorrect/not available address.
                     // This approach would solve the issue...
                 }
             }
 
-            private NetMQFrame ReceiveUdpFrame(out string peerName)
+            private bool TryReceiveUdpFrame(out NetMQFrame frame, out string peerName)
             {
                 Assumes.NotNull(m_udpSocket);
 
                 var buffer = new byte[UdpFrameMax];
                 EndPoint peer = new IPEndPoint(IPAddress.Any, 0);
 
-                var bytesRead = m_udpSocket.ReceiveFrom(buffer, ref peer);
-                peerName = peer.ToString();
+                int bytesRead = 0;
+                try
+                {
+                    bytesRead = m_udpSocket.ReceiveFrom(buffer, ref peer);
+                }
+                catch (SocketException ex) when (ex.SocketErrorCode == SocketError.MessageSize)
+                {
+                    frame = default;
+                    peerName = null;
+                    return false;
+                }
 
-                return new NetMQFrame(buffer, bytesRead);
+                peerName = peer.ToString();
+                frame = new NetMQFrame(buffer, bytesRead);
+                return true;
             }
         }