3party/libzmq
0
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-03-17 08:34:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

Couple of bugs in XREP handling of identities fixed.

Browse Source 
wq:
Signed-off-by: Martin Sustrik <sustrik@250bpm.com>
This commit is contained in:
Martin Sustrik 2011-11-13 10:33:49 +01:00
parent f8b0055026
commit 1c239708ab
1 changed files with 21 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
src/xrep.cpp
View File

@ -189,14 +189,18 @@ int zmq::xrep_t::xrecv (msg_t *msg_, int flags_)
return 0; return 0;
} }
// Get next message part. pipe_t *pipe = NULL;
pipe_t *pipe; while (true) {
int rc = fq.recvpipe (msg_, flags_, &pipe);
if (rc != 0) // Get next message part.
return -1; int rc = fq.recvpipe (msg_, flags_, &pipe);
if (rc != 0)
return -1;
// If identity is received, change the key assigned to the pipe.
if (likely (!(msg_->flags () & msg_t::identity)))
break;
// If identity is received, change the key assigned to the pipe.
if (unlikely (msg_->flags () & msg_t::identity)) {
zmq_assert (!more_in); zmq_assert (!more_in);
// Empty identity means we can preserve the auto-generated identity. // Empty identity means we can preserve the auto-generated identity.
@ -219,11 +223,6 @@ int zmq::xrep_t::xrecv (msg_t *msg_, int flags_)
} }
zmq_assert (it != outpipes.end ()); zmq_assert (it != outpipes.end ());
} }
// After processing the identity, try to get the next message.
rc = fq.recvpipe (msg_, flags_, &pipe);
if (rc != 0)
return -1;
} }
// If we are in the middle of reading a message, just return the next part. // If we are in the middle of reading a message, just return the next part.
@ -234,7 +233,7 @@ int zmq::xrep_t::xrecv (msg_t *msg_, int flags_)
// We are at the beginning of a new message. Move the message part we // We are at the beginning of a new message. Move the message part we
// have to the prefetched and return the ID of the peer instead. // have to the prefetched and return the ID of the peer instead.
rc = prefetched_msg.move (*msg_); int rc = prefetched_msg.move (*msg_);
errno_assert (rc == 0); errno_assert (rc == 0);
prefetched = true; prefetched = true;
rc = msg_->close (); rc = msg_->close ();
@ -260,9 +259,17 @@ int zmq::xrep_t::rollback (void)
bool zmq::xrep_t::xhas_in () bool zmq::xrep_t::xhas_in ()
{ {
// We may already have a message pre-fetched.
if (prefetched) if (prefetched)
return true; return true;
return fq.has_in ();
// 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::xrep_t::xhas_out () bool zmq::xrep_t::xhas_out ()