summaryrefslogtreecommitdiff
path: root/src/xreq.cpp
diff options
context:
space:
mode:
authorMartin Sustrik <sustrik@250bpm.com>2011-11-14 11:15:20 +0100
committerMartin Sustrik <sustrik@250bpm.com>2011-11-14 11:15:20 +0100
commit21bca4dbe48cffdaa1e1108600973559a3b880d2 (patch)
tree7293029a23b8cb32488a5e25be1b1ffb382c5587 /src/xreq.cpp
parent1c239708ab174c1de9f99e256d23158f74a24dbc (diff)
Bug concerining identity in XREQ socket fixed (issue 280)
Signed-off-by: Martin Sustrik <sustrik@250bpm.com>
Diffstat (limited to 'src/xreq.cpp')
-rw-r--r--src/xreq.cpp26
1 files changed, 24 insertions, 2 deletions
diff --git a/src/xreq.cpp b/src/xreq.cpp
index 91317f7..59d3f39 100644
--- a/src/xreq.cpp
+++ b/src/xreq.cpp
@@ -24,7 +24,8 @@
#include "msg.hpp"
zmq::xreq_t::xreq_t (class ctx_t *parent_, uint32_t tid_) :
- socket_base_t (parent_, tid_)
+ socket_base_t (parent_, tid_),
+ prefetched (false)
{
options.type = ZMQ_XREQ;
@@ -36,10 +37,13 @@ zmq::xreq_t::xreq_t (class ctx_t *parent_, uint32_t tid_) :
options.send_identity = true;
options.recv_identity = true;
+
+ prefetched_msg.init ();
}
zmq::xreq_t::~xreq_t ()
{
+ prefetched_msg.close ();
}
void zmq::xreq_t::xattach_pipe (pipe_t *pipe_)
@@ -56,6 +60,14 @@ int zmq::xreq_t::xsend (msg_t *msg_, int flags_)
int zmq::xreq_t::xrecv (msg_t *msg_, int flags_)
{
+ // If there is a prefetched message, return it.
+ if (prefetched) {
+ int rc = msg_->move (prefetched_msg);
+ errno_assert (rc == 0);
+ prefetched = false;
+ return 0;
+ }
+
// XREQ socket doesn't use identities. We can safely drop it and
while (true) {
int rc = fq.recv (msg_, flags_);
@@ -69,7 +81,17 @@ int zmq::xreq_t::xrecv (msg_t *msg_, int flags_)
bool zmq::xreq_t::xhas_in ()
{
- return fq.has_in ();
+ // We may already have a message pre-fetched.
+ if (prefetched)
+ return true;
+
+ // Try to read the next message to the pre-fetch buffer.
+ int rc = xrecv (&prefetched_msg, ZMQ_DONTWAIT);
+ if (rc != 0 && errno == EAGAIN)
+ return false;
+ zmq_assert (rc == 0);
+ prefetched = true;
+ return true;
}
bool zmq::xreq_t::xhas_out ()