generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-10-27 19:10:22 +01:00
Code Issues Projects Releases Wiki Activity

ROUTER socket blocks on SNDHWM

Browse Source 
Till now the message was droppen in such case.

Signed-off-by: Martin Sustrik <sustrik@250bpm.com>
This commit is contained in:
Martin Sustrik
2011-07-21 19:12:51 +02:00
parent a1e09facb2
commit 6b873d4ffd
1 changed files with 36 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
src/router.cpp
View File

@@ -129,14 +129,13 @@ int zmq::router_t::xsend (msg_t *msg_, int flags_)
if (!more_out) { if (!more_out) {
zmq_assert (!current_out); zmq_assert (!current_out);
// If we have malformed message (prefix with no subsequent message) // The first message part has to be label.
// then just silently ignore it. if (unlikely (!(msg_->flags () & msg_t::label))) {
// TODO: The connections should be killed instead. errno = EFSM;
if (msg_->flags () & msg_t::label) { return -1;
}
more_out = true; // Find the pipe associated with the peer ID stored in the message.
// Find the pipe associated with the peer ID stored in the prefix.
if (unlikely (msg_->size () != 4)) { if (unlikely (msg_->size () != 4)) {
errno = ECANTROUTE; errno = ECANTROUTE;
return -1; return -1;
@@ -149,20 +148,25 @@ int zmq::router_t::xsend (msg_t *msg_, int flags_)
} }
// Check whether the pipe is available for writing. // Check whether the pipe is available for writing.
current_out = it->second.pipe;
msg_t empty; msg_t empty;
int rc = empty.init (); int rc = empty.init ();
errno_assert (rc == 0); errno_assert (rc == 0);
if (!current_out->check_write (&empty)) { if (!it->second.pipe->check_write (&empty)) {
rc = empty.close ();
errno_assert (rc == 0);
it->second.active = false; it->second.active = false;
more_out = false; errno = EAGAIN;
current_out = NULL; return -1;
} }
rc = empty.close (); rc = empty.close ();
errno_assert (rc == 0); errno_assert (rc == 0);
}
int rc = msg_->close (); // Mark the pipe to send the message to.
current_out = it->second.pipe;
more_out = true;
// Clean up the message object.
rc = msg_->close ();
errno_assert (rc == 0); errno_assert (rc == 0);
rc = msg_->init (); rc = msg_->init ();
errno_assert (rc == 0); errno_assert (rc == 0);