New wire format for REQ/REP pattern

This patch introduces two changes:
1. 32-bit ID is used to identify the peer instead of UUID
2. REQ socket seeds the label stack with unique 32-bit request ID
   It also drops any replies with non-matching request ID

Signed-off-by: Martin Sustrik <sustrik@250bpm.com>

1 files changed, 33 insertions, 5 deletions

diff --git a/src/req.cpp b/src/req.cpp
index b0e58dc..e51b853 100644
--- a/src/req.cpp
+++ b/src/req.cpp
@@ -21,13 +21,21 @@
 #include "req.hpp"
 #include "err.hpp"
 #include "msg.hpp"
+#include "uuid.hpp"
+#include "wire.hpp"
+#include "random.hpp"
+#include "likely.hpp"
 
 zmq::req_t::req_t (class ctx_t *parent_, uint32_t tid_) :
     xreq_t (parent_, tid_),
     receiving_reply (false),
-    message_begins (true)
+    message_begins (true),
+    request_id (0)
 {
     options.type = ZMQ_REQ;
+
+    //  Start the request ID sequence at an random point.
+    generate_random (&request_id, sizeof (request_id));
 }
 
 zmq::req_t::~req_t ()
@@ -43,12 +51,14 @@ int zmq::req_t::xsend (msg_t *msg_, int flags_)
         return -1;
     }
 
-    //  First part of the request is empty message part (stack bottom).
+    //  First part of the request is the request identity.
     if (message_begins) {
         msg_t prefix;
-        int rc = prefix.init ();
+        int rc = prefix.init_size (4);
         errno_assert (rc == 0);
         prefix.set_flags (msg_t::label);
+        unsigned char *data = (unsigned char*) prefix.data ();
+        put_uint32 (data, request_id);
         rc = xreq_t::xsend (&prefix, flags_);
         if (rc != 0)
             return rc;
@@ -78,13 +88,28 @@ int zmq::req_t::xrecv (msg_t *msg_, int flags_)
         return -1;
     }
 
-    //  First part of the reply should be empty message part (stack bottom).
+    //  First part of the reply should be the original request ID.
     if (message_begins) {
         int rc = xreq_t::xrecv (msg_, flags_);
         if (rc != 0)
             return rc;
         zmq_assert (msg_->flags () & msg_t::label);
-        zmq_assert (msg_->size () == 0);
+        zmq_assert (msg_->size () == 4);
+        unsigned char *data = (unsigned char*) msg_->data ();
+        if (unlikely (get_uint32 (data) != request_id)) {
+
+            //  The request ID does not match. Drop the entire message.
+            while (true) {
+                int rc = xreq_t::xrecv (msg_, flags_);
+                errno_assert (rc == 0);
+                if (!(msg_->flags () & (msg_t::label | msg_t::more)))
+                    break;
+            }
+            msg_->close ();
+            msg_->init ();
+            errno = EAGAIN;
+            return -1;
+        }
         message_begins = false;
     }
 
@@ -94,6 +119,7 @@ int zmq::req_t::xrecv (msg_t *msg_, int flags_)
 
     //  If the reply is fully received, flip the FSM into request-sending state.
     if (!(msg_->flags () & (msg_t::more | msg_t::label))) {
+        request_id++;
         receiving_reply = false;
         message_begins = true;
     }
@@ -103,6 +129,8 @@ int zmq::req_t::xrecv (msg_t *msg_, int flags_)
 
 bool zmq::req_t::xhas_in ()
 {
+    //  TODO: Duplicates should be removed here.
+
     if (!receiving_reply)
         return false;