3party/libzmq
0
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2024-12-27 15:41:05 +08:00
Code Issues Projects Releases Wiki Activity

returns EHOSTUNREACH when a peer is full if ZMQ_ROUTER_MANDATORY is set

Browse Source
This commit is contained in:
Min(Dongmin Yu) 2012-12-08 10:20:42 +09:00
parent 95d36f42ee
commit 9382941adc
2 changed files with 52 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/router.cpp
View File

@ -156,19 +156,24 @@ int zmq::router_t::xsend (msg_t *msg_)
// Find the pipe associated with the identity stored in the prefix. // Find the pipe associated with the identity stored in the prefix.
// If there's no such pipe just silently ignore the message, unless // If there's no such pipe just silently ignore the message, unless
// report_unreachable is set. // router_mandatory is set.
blob_t identity ((unsigned char*) msg_->data (), msg_->size ()); blob_t identity ((unsigned char*) msg_->data (), msg_->size ());
outpipes_t::iterator it = outpipes.find (identity); outpipes_t::iterator it = outpipes.find (identity);
bool unreach = false;
if (it != outpipes.end ()) { if (it != outpipes.end ()) {
current_out = it->second.pipe; current_out = it->second.pipe;
if (!current_out->check_write ()) { if (!current_out->check_write ()) {
it->second.active = false; it->second.active = false;
current_out = NULL; current_out = NULL;
unreach = true;
} }
} }
else else
if (mandatory) { if (mandatory)
unreach = true;
if (unreach) {
more_out = false; more_out = false;
errno = EHOSTUNREACH; errno = EHOSTUNREACH;
return -1; return -1;

46
tests/test_router_mandatory.cpp
View File

@ -21,6 +21,7 @@
#include <stdio.h> #include <stdio.h>
#include "testutil.hpp" #include "testutil.hpp"
#include "../include/zmq_utils.h"
int main (void) int main (void)
{ {
@ -33,7 +34,11 @@ int main (void)
void *sa = zmq_socket (ctx, ZMQ_ROUTER); void *sa = zmq_socket (ctx, ZMQ_ROUTER);
assert (sa); assert (sa);
int rc = zmq_bind (sa, "tcp://127.0.0.1:15560"); int hwm = 1;
int rc = zmq_setsockopt (sa, ZMQ_SNDHWM, &hwm, sizeof (hwm));
assert (rc == 0);
rc = zmq_bind (sa, "tcp://127.0.0.1:15560");
assert (rc == 0); assert (rc == 0);
// Sending a message to an unknown peer with the default setting // Sending a message to an unknown peer with the default setting
@ -52,9 +57,48 @@ int main (void)
rc = zmq_send (sa, "UNKNOWN", 7, ZMQ_SNDMORE | ZMQ_DONTWAIT); rc = zmq_send (sa, "UNKNOWN", 7, ZMQ_SNDMORE | ZMQ_DONTWAIT);
assert (rc == -1 && errno == EHOSTUNREACH); assert (rc == -1 && errno == EHOSTUNREACH);
// Create a valid socket
void *sb = zmq_socket (ctx, ZMQ_DEALER);
assert (sb);
rc = zmq_setsockopt (sb, ZMQ_RCVHWM, &hwm, sizeof (hwm));
assert (rc == 0);
rc = zmq_setsockopt (sb, ZMQ_IDENTITY, "X", 1);
assert (rc == 0);
rc = zmq_connect (sb, "tcp://127.0.0.1:15560");
assert (rc == 0);
// wait until connect
zmq_sleep (1);
// make it full and check that it fails
rc = zmq_send (sa, "X", 1, ZMQ_SNDMORE);
assert (rc == 1);
rc = zmq_send (sa, "DATA1", 5, 0);
assert (rc == 5);
rc = zmq_send (sa, "X", 1, ZMQ_SNDMORE);
if (rc == 1) {
// the first frame has been sent
rc = zmq_send (sa, "DATA2", 5, 0);
assert (rc == 5);
// send more
rc = zmq_send (sa, "X", 1, ZMQ_SNDMORE);
}
assert (rc == -1 && errno == EHOSTUNREACH);
rc = zmq_close (sa); rc = zmq_close (sa);
assert (rc == 0); assert (rc == 0);
rc = zmq_close (sb);
assert (rc == 0);
rc = zmq_term (ctx); rc = zmq_term (ctx);
assert (rc == 0); assert (rc == 0);