summaryrefslogtreecommitdiff
path: root/src/req.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/req.cpp')
-rw-r--r--src/req.cpp38
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;