diff options
author | Martin Sustrik <sustrik@250bpm.com> | 2011-06-22 11:02:16 +0200 |
---|---|---|
committer | Martin Sustrik <sustrik@250bpm.com> | 2011-06-22 11:02:16 +0200 |
commit | ec81f8fb2523e1e2fe45eaadc05311a35bf551d7 (patch) | |
tree | e6fbd9b7a789d72678fa02ca06883de15a932beb /src/req.cpp | |
parent | 10a93bb79fd3d4be1b3ffedfa6785564fbcc082b (diff) |
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>
Diffstat (limited to 'src/req.cpp')
-rw-r--r-- | src/req.cpp | 38 |
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; |