Merge pull request #885 from somdoron/sockets

Radio-Dish and Scatter-gather

Author: somdoron

src/NetMQ.Tests/RadioDish.cs

@@ -0,0 +1,69 @@
+using System;
+using System.Threading;
+using NetMQ.Sockets;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace NetMQ.Tests
+{
+    public class RadioDish
+    {
+        [Fact]
+        public void TestTcp()
+        {
+            using var radio = new RadioSocket();
+            using var dish = new DishSocket();
+            dish.Join("2");
+
+            int port = radio.BindRandomPort("tcp://*");
+            dish.Connect($"tcp://127.0.0.1:{port}");
+            
+            Thread.Sleep(100);
+
+            radio.Send("1", "HELLO");
+            radio.Send("2", "HELLO");
+
+            var (group,msg) = dish.ReceiveString();
+            Assert.Equal("2", group);
+            Assert.Equal("HELLO", msg);
+        }
+        
+        [Fact]
+        public void TestBlocking()
+        {
+            using var radio = new RadioSocket();
+            using var dish = new DishSocket();
+            dish.Join("2");
+
+            radio.Bind("inproc://test-radio-dish");
+            dish.Connect("inproc://test-radio-dish");
+            
+            radio.Send("1", "HELLO");
+            radio.Send("2", "HELLO");
+
+            var (group,msg) = dish.ReceiveString();
+            Assert.Equal("2", group);
+            Assert.Equal("HELLO", msg);
+        }
+        
+        [Fact]
+        public async void TestAsync()
+        {
+            using var radio = new RadioSocket();
+            using var dish = new DishSocket();
+            dish.Join("2");
+
+            int port = radio.BindRandomPort("tcp://*");
+            dish.Connect($"tcp://127.0.0.1:{port}");
+            
+            Thread.Sleep(100);
+
+            await radio.SendAsync("1", "HELLO");
+            await radio.SendAsync("2", "HELLO");
+
+            var (group,msg) = await dish.ReceiveStringAsync();
+            Assert.Equal("2", group);
+            Assert.Equal("HELLO", msg);
+        }
+    }
+}

src/NetMQ.Tests/ScatterGather.cs

@@ -0,0 +1,70 @@
+using System.Threading;
+using NetMQ.Sockets;
+using Xunit;
+
+namespace NetMQ.Tests
+{
+    public class ScatterGather
+    {
+        [Fact]
+        public void TestTcp()
+        {
+            using var scatter = new ScatterSocket();
+            using var gather = new GatherSocket();
+
+            int port = scatter.BindRandomPort("tcp://*");
+            gather.Connect($"tcp://127.0.0.1:{port}");
+            
+            scatter.Send("1");
+            scatter.Send("2");
+
+            var m1 = gather.ReceiveString();
+            Assert.Equal("1", m1);
+            
+            var m2 = gather.ReceiveString();
+            Assert.Equal("2", m2);
+        }
+        
+        [Fact]
+        public void TestBlocking()
+        {
+            using var scatter = new ScatterSocket();
+            using var gather = new GatherSocket();
+            using var gather2 = new GatherSocket();
+
+            scatter.Bind("inproc://test-scatter-gather");
+            gather.Connect("inproc://test-scatter-gather");
+            gather2.Connect("inproc://test-scatter-gather");
+            
+            scatter.Send("1");
+            scatter.Send("2");
+
+            var m1 = gather.ReceiveString();
+            Assert.Equal("1", m1);
+            
+            var m2 = gather2.ReceiveString();
+            Assert.Equal("2", m2);
+        }
+        
+        [Fact]
+        public async void TestAsync()
+        {
+            using var scatter = new ScatterSocket();
+            using var gather = new GatherSocket();
+            using var gather2 = new GatherSocket();
+
+            scatter.Bind("inproc://test-scatter-gather");
+            gather.Connect("inproc://test-scatter-gather");
+            gather2.Connect("inproc://test-scatter-gather");
+            
+            await scatter.SendAsync("1");
+            await scatter.SendAsync("2");
+
+            var m1 = await gather.ReceiveStringAsync();
+            Assert.Equal("1", m1);
+            
+            var m2 = await gather2.ReceiveStringAsync();
+            Assert.Equal("2", m2);
+        }
+    }
+}

src/NetMQ/Core/Mechanisms/Mechanism.cs

@@ -33,6 +33,10 @@ static class SocketNames
             public const string Peer = "PEER";
             public const string Server = "SERVER";
             public const string Client = "CLIENT";
+            public const string Radio = "RADIO";
+            public const string Dish = "DISH";
+            public const string Gather = "GATHER";
+            public const string Scatter = "SCATTER";
         }
 
         const int NameLengthSize = sizeof(byte);
@@ -125,6 +129,14 @@ protected string GetSocketName(ZmqSocketType socketType)
                     return SocketNames.Server;
                 case ZmqSocketType.Client:
                     return SocketNames.Client;
+                case ZmqSocketType.Radio:
+                    return SocketNames.Radio;
+                case ZmqSocketType.Dish:
+                    return SocketNames.Dish;
+                case ZmqSocketType.Gather:
+                    return SocketNames.Gather;
+                case ZmqSocketType.Scatter:
+                    return SocketNames.Scatter;
                 default:
                     throw new ArgumentOutOfRangeException(nameof(socketType), socketType, null);
             }
@@ -312,6 +324,14 @@ private bool CheckSocketType(string type)
                     return type == SocketNames.Server;
                 case ZmqSocketType.Peer:
                     return type == SocketNames.Peer;
+                case ZmqSocketType.Radio:
+                    return type == SocketNames.Dish;
+                case ZmqSocketType.Dish:
+                    return type == SocketNames.Radio;
+                case ZmqSocketType.Gather:
+                    return type == SocketNames.Scatter;
+                case ZmqSocketType.Scatter:
+                    return type == SocketNames.Gather;
                 default:
                     return false;
             }