mirror of
https://github.com/zeromq/libzmq.git
synced 2025-10-28 19:52:00 +01:00
Shorten ZMQ_ROUTER_REASSIGN_IDENTITIES to ZMQ_ROUTER_HANDOVER. Also, added a test for the HANDOVER functionality.
This commit is contained in:
@@ -292,7 +292,7 @@ ZMQ_EXPORT int zmq_msg_set (zmq_msg_t *msg, int option, int optval);
|
|||||||
#define ZMQ_REQ_RELAXED 53
|
#define ZMQ_REQ_RELAXED 53
|
||||||
#define ZMQ_CONFLATE 54
|
#define ZMQ_CONFLATE 54
|
||||||
#define ZMQ_ZAP_DOMAIN 55
|
#define ZMQ_ZAP_DOMAIN 55
|
||||||
#define ZMQ_ROUTER_REASSIGN_IDENTITES 56
|
#define ZMQ_ROUTER_HANDOVER 56
|
||||||
|
|
||||||
/* Message options */
|
/* Message options */
|
||||||
#define ZMQ_MORE 1
|
#define ZMQ_MORE 1
|
||||||
|
|||||||
@@ -36,7 +36,7 @@ zmq::router_t::router_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
|
|||||||
// raw_sock functionality in ROUTER is deprecated
|
// raw_sock functionality in ROUTER is deprecated
|
||||||
raw_sock (false),
|
raw_sock (false),
|
||||||
probe_router (false),
|
probe_router (false),
|
||||||
reassign_identities(false)
|
handover(false)
|
||||||
{
|
{
|
||||||
options.type = ZMQ_ROUTER;
|
options.type = ZMQ_ROUTER;
|
||||||
options.recv_identity = true;
|
options.recv_identity = true;
|
||||||
@@ -112,9 +112,9 @@ int zmq::router_t::xsetsockopt (int option_, const void *optval_,
|
|||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
break;
|
break;
|
||||||
case ZMQ_ROUTER_REASSIGN_IDENTITES:
|
case ZMQ_ROUTER_HANDOVER:
|
||||||
if (is_int && value >= 0) {
|
if (is_int && value >= 0) {
|
||||||
reassign_identities = (value != 0);
|
handover = (value != 0);
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
break;
|
break;
|
||||||
@@ -403,7 +403,7 @@ bool zmq::router_t::identify_peer (pipe_t *pipe_)
|
|||||||
|
|
||||||
if (it != outpipes.end ())
|
if (it != outpipes.end ())
|
||||||
{
|
{
|
||||||
if (!reassign_identities) {
|
if (!handover) {
|
||||||
// Ignore peers with duplicate ID.
|
// Ignore peers with duplicate ID.
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -118,7 +118,7 @@ namespace zmq
|
|||||||
// If true, the router will reassign an identity upon encountering a
|
// If true, the router will reassign an identity upon encountering a
|
||||||
// name collision. The new pipe will take the identity, the old pipe
|
// name collision. The new pipe will take the identity, the old pipe
|
||||||
// will be terminated.
|
// will be terminated.
|
||||||
bool reassign_identities;
|
bool handover;
|
||||||
|
|
||||||
router_t (const router_t&);
|
router_t (const router_t&);
|
||||||
const router_t &operator = (const router_t&);
|
const router_t &operator = (const router_t&);
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ noinst_PROGRAMS = test_system \
|
|||||||
test_linger \
|
test_linger \
|
||||||
test_monitor \
|
test_monitor \
|
||||||
test_router_mandatory \
|
test_router_mandatory \
|
||||||
|
test_router_handover \
|
||||||
test_probe_router \
|
test_probe_router \
|
||||||
test_stream \
|
test_stream \
|
||||||
test_stream_empty \
|
test_stream_empty \
|
||||||
@@ -68,6 +69,7 @@ test_last_endpoint_SOURCES = test_last_endpoint.cpp
|
|||||||
test_term_endpoint_SOURCES = test_term_endpoint.cpp
|
test_term_endpoint_SOURCES = test_term_endpoint.cpp
|
||||||
test_monitor_SOURCES = test_monitor.cpp
|
test_monitor_SOURCES = test_monitor.cpp
|
||||||
test_router_mandatory_SOURCES = test_router_mandatory.cpp
|
test_router_mandatory_SOURCES = test_router_mandatory.cpp
|
||||||
|
test_router_handover_SOURCES = test_router_handover.cpp
|
||||||
test_probe_router_SOURCES = test_probe_router.cpp
|
test_probe_router_SOURCES = test_probe_router.cpp
|
||||||
test_stream_SOURCES = test_stream.cpp
|
test_stream_SOURCES = test_stream.cpp
|
||||||
test_stream_empty_SOURCES = test_stream_empty.cpp
|
test_stream_empty_SOURCES = test_stream_empty.cpp
|
||||||
|
|||||||
101
tests/test_router_handover.cpp
Normal file
101
tests/test_router_handover.cpp
Normal file
@@ -0,0 +1,101 @@
|
|||||||
|
/*
|
||||||
|
Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file
|
||||||
|
|
||||||
|
This file is part of 0MQ.
|
||||||
|
|
||||||
|
0MQ is free software; you can redistribute it and/or modify it under
|
||||||
|
the terms of the GNU Lesser General Public License as published by
|
||||||
|
the Free Software Foundation; either version 3 of the License, or
|
||||||
|
(at your option) any later version.
|
||||||
|
|
||||||
|
0MQ is distributed in the hope that it will be useful,
|
||||||
|
but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||||
|
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||||
|
GNU Lesser General Public License for more details.
|
||||||
|
|
||||||
|
You should have received a copy of the GNU Lesser General Public License
|
||||||
|
along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||||
|
*/
|
||||||
|
|
||||||
|
#include "testutil.hpp"
|
||||||
|
|
||||||
|
int main (void)
|
||||||
|
{
|
||||||
|
setup_test_environment();
|
||||||
|
void *ctx = zmq_ctx_new ();
|
||||||
|
assert (ctx);
|
||||||
|
void *router = zmq_socket (ctx, ZMQ_ROUTER);
|
||||||
|
assert (router);
|
||||||
|
|
||||||
|
int rc = zmq_bind (router, "tcp://127.0.0.1:5560");
|
||||||
|
assert (rc == 0);
|
||||||
|
|
||||||
|
// Enable the handover flag
|
||||||
|
int handover = 1;
|
||||||
|
rc = zmq_setsockopt (router, ZMQ_ROUTER_HANDOVER, &handover, sizeof (handover));
|
||||||
|
assert (rc == 0);
|
||||||
|
|
||||||
|
// Create dealer called "X" and connect it to our router
|
||||||
|
void *dealer_one = zmq_socket (ctx, ZMQ_DEALER);
|
||||||
|
assert (dealer_one);
|
||||||
|
rc = zmq_setsockopt (dealer_one, ZMQ_IDENTITY, "X", 1);
|
||||||
|
assert (rc == 0);
|
||||||
|
rc = zmq_connect (dealer_one, "tcp://127.0.0.1:5560");
|
||||||
|
assert (rc == 0);
|
||||||
|
|
||||||
|
// Get message from dealer to know when connection is ready
|
||||||
|
char buffer [255];
|
||||||
|
rc = zmq_send (dealer_one, "Hello", 5, 0);
|
||||||
|
assert (rc == 5);
|
||||||
|
rc = zmq_recv (router, buffer, 255, 0);
|
||||||
|
assert (rc == 1);
|
||||||
|
assert (buffer [0] == 'X');
|
||||||
|
rc = zmq_recv (router, buffer, 255, 0);
|
||||||
|
assert (rc == 5);
|
||||||
|
|
||||||
|
// Now create a second dealer that uses the same identity
|
||||||
|
void *dealer_two = zmq_socket (ctx, ZMQ_DEALER);
|
||||||
|
assert (dealer_two);
|
||||||
|
rc = zmq_setsockopt (dealer_two, ZMQ_IDENTITY, "X", 1);
|
||||||
|
assert (rc == 0);
|
||||||
|
rc = zmq_connect (dealer_two, "tcp://127.0.0.1:5560");
|
||||||
|
assert (rc == 0);
|
||||||
|
|
||||||
|
// Get message from dealer to know when connection is ready
|
||||||
|
rc = zmq_send (dealer_two, "Hello", 5, 0);
|
||||||
|
assert (rc == 5);
|
||||||
|
rc = zmq_recv (router, buffer, 255, 0);
|
||||||
|
assert (rc == 1);
|
||||||
|
assert (buffer [0] == 'X');
|
||||||
|
rc = zmq_recv (router, buffer, 255, 0);
|
||||||
|
assert (rc == 5);
|
||||||
|
|
||||||
|
// Send a message to 'X' identity. This should be delivered
|
||||||
|
// to the second dealer, instead of the first beccause of the handover.
|
||||||
|
rc = zmq_send (router, "X", 1, ZMQ_SNDMORE);
|
||||||
|
assert (rc == 1);
|
||||||
|
rc = zmq_send (router, "Hello", 5, 0);
|
||||||
|
assert (rc == 5);
|
||||||
|
|
||||||
|
// Ensure that the first dealer doesn't receive the message
|
||||||
|
// but the second one does
|
||||||
|
rc = zmq_recv (dealer_one, buffer, 255, ZMQ_NOBLOCK);
|
||||||
|
assert (rc == -1);
|
||||||
|
|
||||||
|
rc = zmq_recv (dealer_two, buffer, 255, 0);
|
||||||
|
assert (rc == 5);
|
||||||
|
|
||||||
|
rc = zmq_close (router);
|
||||||
|
assert (rc == 0);
|
||||||
|
|
||||||
|
rc = zmq_close (dealer_one);
|
||||||
|
assert (rc == 0);
|
||||||
|
|
||||||
|
rc = zmq_close (dealer_two);
|
||||||
|
assert (rc == 0);
|
||||||
|
|
||||||
|
rc = zmq_ctx_term (ctx);
|
||||||
|
assert (rc == 0);
|
||||||
|
|
||||||
|
return 0 ;
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user