Merge pull request #869 from dhyams/relaxed_correlate_866

Added support for ZMQ_REQ_CORRELATE and ZMQ_REQ_RELAXED socket options

Author: somdoron

src/NetMQ.Tests/ReqRepTests.cs

@@ -7,50 +7,131 @@ public class ReqRepTests : IClassFixture<CleanupAfterFixture>
     {
         public ReqRepTests() => NetMQConfig.Cleanup();
 
+        protected void SimpleReqRepSequence(string address, RequestSocket req, ResponseSocket rep)
+        {
+            var port = rep.BindRandomPort(address);
+            req.Connect(address + ":" + port);
+
+            req.SendFrame("Hi");
+            Assert.Equal(new[] { "Hi" }, rep.ReceiveMultipartStrings());
+            rep.SendFrame("Hi2");
+            Assert.Equal(new[] { "Hi2" }, req.ReceiveMultipartStrings());
+        }
+
+        protected void SendTwoReqsInSuccession(string address, RequestSocket req, ResponseSocket rep)
+        {
+            var port = rep.BindRandomPort(address);
+            req.Connect(address + ":" + port);
+            req.SendFrame("Hi");
+            rep.SkipFrame();
+        }
+
+        [Theory]
+        [InlineData("tcp://localhost")]
+        [InlineData("tcp://127.0.0.1")]
+        public void SimpleReqRepSucceeds(string address)
+        {
+            using (var rep = new ResponseSocket())
+            using (var req = new RequestSocket())
+            {
+                SimpleReqRepSequence(address, req, rep);
+            }
+        }
+
+        [Theory]
+        [InlineData("tcp://localhost")]
+        [InlineData("tcp://127.0.0.1")]
+        public void SimpleReqRepWithCorrelationSucceeds(string address)
+        {
+            using (var rep = new ResponseSocket())
+            using (var req = new RequestSocket())
+            {
+                req.Options.Correlate = true;
+                SimpleReqRepSequence(address, req, rep);
+            }
+        }
+
         [Theory]
         [InlineData("tcp://localhost")]
         [InlineData("tcp://127.0.0.1")]
-        public void SimpleReqRep(string address)
+        public void SimpleReqRepWithRelaxedSucceeds(string address)
+        {
+            using (var rep = new ResponseSocket())
+            using (var req = new RequestSocket())
+            {
+                req.Options.Relaxed = true;
+                SimpleReqRepSequence(address, req, rep);
+                req.SendFrame("Hi2"); // ick that this not what I wanted.
+            }
+        }
+
+        [Theory]
+        [InlineData("tcp://localhost")]
+        [InlineData("tcp://127.0.0.1")]
+        public void SendingTwoRequestsInSuccessionFails(string address)
         {
             using (var rep = new ResponseSocket())
             using (var req = new RequestSocket())
             {
                 var port = rep.BindRandomPort(address);
                 req.Connect(address + ":" + port);
-
                 req.SendFrame("Hi");
+                rep.SkipFrame();
+                Assert.Throws<FiniteStateMachineException>(() => req.SendFrame("Hi2"));
+            }
+        }
 
-                Assert.Equal(new[] { "Hi" }, rep.ReceiveMultipartStrings());
-
-                rep.SendFrame("Hi2");
+        [Theory]
+        [InlineData("tcp://localhost")]
+        [InlineData("tcp://127.0.0.1")]
+        public void SendingTwoRequestsInSuccessionWithRelaxedSucceeds(string address)
+        {
+            using (var rep = new ResponseSocket())
+            using (var req = new RequestSocket())
+            {
+                req.Options.Relaxed = true;
 
-                Assert.Equal(new[] { "Hi2" }, req.ReceiveMultipartStrings());
+                var port = rep.BindRandomPort(address);
+                req.Connect(address + ":" + port);
+                req.SendFrame("Hi");
+                rep.SkipFrame();
             }
         }
 
-        [Fact]
-        public void SendingTwoRequestsInARow()
+        [Fact] 
+        public void ReceiveBeforeSendingFails()
         {
             using (var rep = new ResponseSocket())
             using (var req = new RequestSocket())
             {
                 var port = rep.BindRandomPort("tcp://localhost");
                 req.Connect("tcp://localhost:" + port);
 
-                req.SendFrame("Hi");
+                Assert.Throws<FiniteStateMachineException>(() => req.ReceiveFrameBytes());
+            }
+        }
 
-                rep.SkipFrame();
+        [Fact]
+        public void ReceiveBeforeSendingWithRelaxedStillFails()
+        {
+            using (var rep = new ResponseSocket())
+            using (var req = new RequestSocket())
+            {
+                req.Options.Relaxed = true;
+                var port = rep.BindRandomPort("tcp://localhost");
+                req.Connect("tcp://localhost:" + port);
 
-                Assert.Throws<FiniteStateMachineException>(() => req.SendFrame("Hi2"));
+                Assert.Throws<FiniteStateMachineException>(() => req.ReceiveFrameBytes());
             }
         }
 
         [Fact]
-        public void ReceiveBeforeSending()
+        public void ReceiveBeforeSendingWithCorrelateStillFails()
         {
             using (var rep = new ResponseSocket())
             using (var req = new RequestSocket())
             {
+                req.Options.Correlate = true;
                 var port = rep.BindRandomPort("tcp://localhost");
                 req.Connect("tcp://localhost:" + port);
 
@@ -59,7 +140,7 @@ public void ReceiveBeforeSending()
         }
 
         [Fact]
-        public void SendMessageInResponseBeforeReceiving()
+        public void SendMessageInResponseBeforeReceivingFails()
         {
             using (var rep = new ResponseSocket())
             using (var req = new RequestSocket())
@@ -71,8 +152,89 @@ public void SendMessageInResponseBeforeReceiving()
             }
         }
 
+        // make sure that a single responder sends messages back to the correct requestors.
+        [Fact]
+        public void SingleResponderSendsCorrectMessagesToMultipleRequestors()
+        {
+            using (var rep = new ResponseSocket())
+            using (var req1 = new RequestSocket())
+            using (var req2 = new RequestSocket())
+            {
+                var port = rep.BindRandomPort("tcp://127.0.0.1");
+
+                req1.Connect($"tcp://127.0.0.1:{port}");
+                req2.Connect($"tcp://127.0.0.1:{port}");
+
+                req1.SendFrame("From1");
+                req2.SendFrame("From2");
+
+                rep.SendFrame(rep.ReceiveFrameString());
+                rep.SendFrame(rep.ReceiveFrameString());
+
+                Assert.Equal("From2", req2.ReceiveFrameString());
+                Assert.Equal("From1", req1.ReceiveFrameString());
+            }
+        }
+
+        internal void RouterBounce(ref RouterSocket router)
+        {
+            bool more;
+            do
+            {
+                var bytes = router.ReceiveFrameBytes(out more);
+                router.SendFrame(bytes, more);
+            } while (more);
+        }
+
+        [Theory]
+        [InlineData(false)]
+        [InlineData(true)]
+        public void CorrelationSelectsLatestRequest(bool correlate)
+        {
+            var rep = new RouterSocket();
+            var req = new RequestSocket();
+
+            var port = rep.BindRandomPort("tcp://localhost");
+            req.Connect($"tcp://localhost:{port}");
+                
+            req.Options.Relaxed = true;
+            req.Options.Correlate = correlate;
+
+            //  Send two requests.
+            req.SendFrame("FirstReq");
+            req.SendFrame("SecondReq");
+
+            //  Bind server allowing it to receive messages.
+            //rep.Bind($"tcp://localhost:{port}");
+
+            //  Read the two messages and send them back as is.
+            RouterBounce(ref rep);
+            RouterBounce(ref rep);
+              
+            //  Read the reply. When Options.Correlate is active,
+            //  "FirstReq" should be ditched and "SecondReq" should be read.  Vice
+            //  versa when Options.Corellate is not active.
+            var result = req.ReceiveFrameString();
+
+            if (correlate)
+            {
+                // if correlate is on, we get SecondReq which is the typical desired behavior; this
+                // is the last request that was sent.
+                Assert.Equal("SecondReq", result); 
+            }
+            else
+            {
+                // if correlate is off, we get FirstReq, which is not desired behavior; this is not
+                // the last request that was sent.
+                Assert.Equal("FirstReq", result); 
+            }
+            
+            rep.Dispose();
+            req.Dispose();
+        }
+
         [Fact]
-        public void SendMultipartMessage()
+        public void SendMultipartMessageSucceeds()
         {
             using (var rep = new ResponseSocket())
             using (var req = new RequestSocket())

src/NetMQ/Core/Options.cs

@@ -66,6 +66,8 @@ public Options()
             HeartbeatTimeout = -1;
             HelloMsg = null;
             CanSendHelloMsg = false;
+            Correlate = false;
+            Relaxed = false;
         }
 
         /// <summary>
@@ -324,7 +326,10 @@ public byte IdentitySize {
         /// Indicate of socket can send an hello msg
         /// </summary>
         public bool CanSendHelloMsg { get; set; }
-        
+
+        public bool Correlate { get; set; }
+        public bool Relaxed { get; set; }
+
         /// <summary>
         /// Assign the given optionValue to the specified option.
         /// </summary>
@@ -520,6 +525,18 @@ public void SetSocketOption(ZmqSocketOption option, object optionValue)
                     }
                     break;
                 }
+
+                case ZmqSocketOption.Relaxed:
+                    {
+                        Relaxed = (bool)optionValue;
+                        break;
+                    }
+
+                case ZmqSocketOption.Correlate:
+                    {
+                        Correlate = (bool)optionValue;
+                        break;
+                    }
 
                 default:
                     throw new InvalidException("Options.SetSocketOption called with invalid ZmqSocketOption of " + option);

src/NetMQ/Core/Patterns/Dealer.cs

@@ -74,6 +74,17 @@ protected override bool XSend(ref Msg msg)
             return m_loadBalancer.Send(ref msg);
         }
 
+        /// <summary>
+        /// Transmit the given message.  The <c>Send</c> method calls this to do the actual sending.
+        /// </summary>
+        /// <param name="msg">the message to transmit</param>
+        /// <param name="pipe">the pipe that the message was transmitted on (output)</param>
+        /// <returns><c>true</c> if the message was sent successfully</returns>
+        protected bool XSendPipe(ref Msg msg, out Pipe pipe)
+        {
+            return m_loadBalancer.SendPipe(ref msg, out pipe);
+        }
+
         /// <summary>
         /// Get a message from FairQueuing data structure
         /// </summary>
@@ -84,6 +95,17 @@ protected override bool XRecv(ref Msg msg)
             return m_fairQueueing.Recv(ref msg);
         }
 
+        /// <summary>
+        /// Get a message from FairQueuing data structure
+        /// </summary>
+        /// <param name="msg">a Msg to receive the message into</param>
+        /// <param name="pipe">a specific Pipe to receive on</param>
+        /// <returns><c>true</c> if the message was received successfully, <c>false</c> if there were no messages to receive</returns>
+        protected bool XRecvPipe(ref Msg msg, out Pipe pipe)
+        {
+            return m_fairQueueing.RecvPipe(ref msg, out pipe);
+        }
+
         /// <summary>
         /// If there is a message available and one has not been pre-fetched yet,
         /// preserve that message as our pre-fetched one.

src/NetMQ/Core/Patterns/Peer.cs

@@ -281,34 +281,32 @@ protected override bool XRecv(ref Msg msg)
                 return true;
             }
 
-            var pipe = new Pipe[1];
-
-            bool isMessageAvailable = m_fairQueueing.RecvPipe(pipe, ref msg);
+            bool isMessageAvailable = m_fairQueueing.RecvPipe(ref msg, out Pipe pipe);
 
             // Drop any messages with more flag
             while (isMessageAvailable && msg.HasMore)
             {
                 // drop all frames of the current multi-frame message
-                isMessageAvailable = m_fairQueueing.RecvPipe(pipe, ref msg);
+                isMessageAvailable = m_fairQueueing.RecvPipe(ref msg, out pipe);
 
                 while (isMessageAvailable && msg.HasMore)
-                    isMessageAvailable = m_fairQueueing.RecvPipe(pipe, ref msg);
+                    isMessageAvailable = m_fairQueueing.RecvPipe(ref msg, out pipe);
 
                 // get the new message
-                isMessageAvailable = m_fairQueueing.RecvPipe(pipe, ref msg);
+                isMessageAvailable = m_fairQueueing.RecvPipe(ref msg, out pipe);
             }
 
             if (!isMessageAvailable)            
                 return false;            
 
-            Debug.Assert(pipe[0] != null);
+            Debug.Assert(pipe != null);
 
             // We are at the beginning of a message.
             // Keep the message part we have in the prefetch buffer
             // and return the ID of the peer instead.
             m_prefetchedMsg.Move(ref msg);
 
-            byte[] routingId = pipe[0].RoutingId;
+            byte[] routingId = pipe.RoutingId;
             msg.InitPool(routingId.Length);
             msg.Put(routingId, 0, routingId.Length);
             msg.SetFlags(MsgFlags.More);