generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2024-12-12 18:40:27 +01:00
Code Issues Projects Releases Wiki Activity

ROUTER socket reports error when message cannot be routed

Browse Source 
Till now, message was silently dropped if it was sent to
a non-existent peer. Now, ECANTROUTE error is returned.

Signed-off-by: Martin Sustrik <sustrik@250bpm.com>
This commit is contained in:
Martin Sustrik 2011-07-21 18:54:27 +02:00
parent 72a793f78a
commit a1e09facb2
3 changed files with 24 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
doc/zmq_send.txt
View File

@ -79,6 +79,9 @@ The provided 'socket' was invalid.
*EINTR*::
The operation was interrupted by delivery of a signal before the message was
sent.
*ECANTROUTE*::
Message cannot be routed to the destination specified as the peer is either
dead or disconnected. This error makes sense only with ZMQ_ROUTER socket.
EXAMPLE

1
include/zmq.h
View File

@ -107,6 +107,7 @@ ZMQ_EXPORT void zmq_version (int *major, int *minor, int *patch);
#define ENOCOMPATPROTO (ZMQ_HAUSNUMERO + 52)
#define ETERM (ZMQ_HAUSNUMERO + 53)
#define EMTHREAD (ZMQ_HAUSNUMERO + 54)
#define ECANTROUTE (ZMQ_HAUSNUMERO + 55)
/* This function retrieves the errno as it is known to 0MQ library. The goal */
/* of this function is to make the code 100% portable, including where 0MQ */

35
src/router.cpp
View File

@ -137,24 +137,29 @@ int zmq::router_t::xsend (msg_t *msg_, int flags_)
more_out = true;
// Find the pipe associated with the peer ID stored in the prefix.
// If there's no such pipe just silently ignore the message.
zmq_assert (msg_->size () == 4);
if (unlikely (msg_->size () != 4)) {
errno = ECANTROUTE;
return -1;
}
uint32_t peer_id = get_uint32 ((unsigned char*) msg_->data ());
outpipes_t::iterator it = outpipes.find (peer_id);
if (it != outpipes.end ()) {
current_out = it->second.pipe;
msg_t empty;
int rc = empty.init ();
errno_assert (rc == 0);
if (!current_out->check_write (&empty)) {
it->second.active = false;
more_out = false;
current_out = NULL;
}
rc = empty.close ();
errno_assert (rc == 0);
if (unlikely (it == outpipes.end ())) {
errno = ECANTROUTE;
return -1;
}
// Check whether the pipe is available for writing.
current_out = it->second.pipe;
msg_t empty;
int rc = empty.init ();
errno_assert (rc == 0);
if (!current_out->check_write (&empty)) {
it->second.active = false;
more_out = false;
current_out = NULL;
}
rc = empty.close ();
errno_assert (rc == 0);
}
int rc = msg_->close ();