mirror of
https://github.com/zeromq/libzmq.git
synced 2024-12-13 10:52:56 +01:00
Issue #1017: add ZMQ_HANDSHAKE_IVL time limit on connection handshake
This commit is contained in:
parent
1cf12ee612
commit
afe5fd87e9
@ -164,6 +164,22 @@ Default value:: N/A
|
||||
Applicable socket types:: all
|
||||
|
||||
|
||||
ZMQ_HANDSHAKE_IVL: Retrieve maximum handshake interval
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
The 'ZMQ_HANDSHAKE_IVL' option shall retrieve the maximum handshake interval
|
||||
for the specified 'socket'. Handshaking is the exchange of socket configuration
|
||||
information (socket type, identity, security) that occurs when a connection
|
||||
is first opened, only for connection-oriented transports. If handshaking does
|
||||
not complete within the configured time, the connection shall be closed.
|
||||
The value 0 means no handshake time limit.
|
||||
|
||||
[horizontal]
|
||||
Option value type:: int
|
||||
Option value unit:: milliseconds
|
||||
Default value:: 30000
|
||||
Applicable socket types:: all but ZMQ_STREAM, only for connection-oriented transports
|
||||
|
||||
|
||||
ZMQ_IDENTITY: Retrieve socket identity
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
The 'ZMQ_IDENTITY' option shall retrieve the identity of the specified 'socket'.
|
||||
|
@ -170,6 +170,22 @@ Default value:: NULL
|
||||
Applicable socket types:: all, when using TCP transport
|
||||
|
||||
|
||||
ZMQ_HANDSHAKE_IVL: Set maximum handshake interval
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
The 'ZMQ_HANDSHAKE_IVL' option shall set the maximum handshake interval for
|
||||
the specified 'socket'. Handshaking is the exchange of socket configuration
|
||||
information (socket type, identity, security) that occurs when a connection
|
||||
is first opened, only for connection-oriented transports. If handshaking does
|
||||
not complete within the configured time, the connection shall be closed.
|
||||
The value 0 means no handshake time limit.
|
||||
|
||||
[horizontal]
|
||||
Option value type:: int
|
||||
Option value unit:: milliseconds
|
||||
Default value:: 30000
|
||||
Applicable socket types:: all but ZMQ_STREAM, only for connection-oriented transports
|
||||
|
||||
|
||||
ZMQ_IDENTITY: Set socket identity
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
The 'ZMQ_IDENTITY' option shall set the identity of the specified 'socket'
|
||||
|
@ -300,6 +300,7 @@ ZMQ_EXPORT const char *zmq_msg_gets (zmq_msg_t *msg, const char *property);
|
||||
#define ZMQ_GSSAPI_PRINCIPAL 63
|
||||
#define ZMQ_GSSAPI_SERVICE_PRINCIPAL 64
|
||||
#define ZMQ_GSSAPI_PLAINTEXT 65
|
||||
#define ZMQ_HANDSHAKE_IVL 66
|
||||
|
||||
/* Message options */
|
||||
#define ZMQ_MORE 1
|
||||
|
@ -55,7 +55,8 @@ zmq::options_t::options_t () :
|
||||
as_server (0),
|
||||
gss_plaintext (false),
|
||||
socket_id (0),
|
||||
conflate (false)
|
||||
conflate (false),
|
||||
handshake_ivl (30000)
|
||||
{
|
||||
}
|
||||
|
||||
@ -435,6 +436,13 @@ int zmq::options_t::setsockopt (int option_, const void *optval_,
|
||||
}
|
||||
break;
|
||||
|
||||
case ZMQ_HANDSHAKE_IVL:
|
||||
if (is_int && value >= 0) {
|
||||
handshake_ivl = value;
|
||||
return 0;
|
||||
}
|
||||
break;
|
||||
|
||||
|
||||
default:
|
||||
break;
|
||||
@ -746,6 +754,13 @@ int zmq::options_t::getsockopt (int option_, void *optval_, size_t *optvallen_)
|
||||
}
|
||||
break;
|
||||
|
||||
case ZMQ_HANDSHAKE_IVL:
|
||||
if (is_int) {
|
||||
*value = handshake_ivl;
|
||||
return 0;
|
||||
}
|
||||
break;
|
||||
|
||||
|
||||
}
|
||||
errno = EINVAL;
|
||||
|
@ -171,6 +171,11 @@ namespace zmq
|
||||
// Cannot receive multi-part messages.
|
||||
// Ignores hwm
|
||||
bool conflate;
|
||||
|
||||
// If connection handshake is not done after this many milliseconds,
|
||||
// close socket. Default is 30 secs. 0 means no handshake timeout.
|
||||
int handshake_ivl;
|
||||
|
||||
};
|
||||
}
|
||||
|
||||
|
@ -80,6 +80,7 @@ zmq::stream_engine_t::stream_engine_t (fd_t fd_, const options_t &options_,
|
||||
mechanism (NULL),
|
||||
input_stopped (false),
|
||||
output_stopped (false),
|
||||
has_handshake_timer (false),
|
||||
socket (NULL)
|
||||
{
|
||||
int rc = tx_msg.init ();
|
||||
@ -197,6 +198,9 @@ void zmq::stream_engine_t::plug (io_thread_t *io_thread_,
|
||||
session->flush ();
|
||||
}
|
||||
else {
|
||||
// start optional timer, to prevent handshake hanging on no input
|
||||
set_handshake_timer ();
|
||||
|
||||
// Send the 'length' and 'flags' fields of the identity message.
|
||||
// The 'length' field is encoded in the long format.
|
||||
outpos = greeting_send;
|
||||
@ -217,6 +221,12 @@ void zmq::stream_engine_t::unplug ()
|
||||
zmq_assert (plugged);
|
||||
plugged = false;
|
||||
|
||||
// Cancel all timers.
|
||||
if (has_handshake_timer) {
|
||||
cancel_timer (handshake_timer_id);
|
||||
has_handshake_timer = false;
|
||||
}
|
||||
|
||||
// Cancel all fd subscriptions.
|
||||
if (!io_error)
|
||||
rm_fd (handle);
|
||||
@ -631,6 +641,11 @@ bool zmq::stream_engine_t::handshake ()
|
||||
// Switch into the normal message flow.
|
||||
handshaking = false;
|
||||
|
||||
if (has_handshake_timer) {
|
||||
cancel_timer (handshake_timer_id);
|
||||
has_handshake_timer = false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
@ -969,3 +984,22 @@ int zmq::stream_engine_t::read (void *data_, size_t size_)
|
||||
|
||||
#endif
|
||||
}
|
||||
|
||||
void zmq::stream_engine_t::set_handshake_timer ()
|
||||
{
|
||||
zmq_assert (!has_handshake_timer);
|
||||
|
||||
if (!options.raw_sock && options.handshake_ivl > 0) {
|
||||
add_timer (options.handshake_ivl, handshake_timer_id);
|
||||
has_handshake_timer = true;
|
||||
}
|
||||
}
|
||||
|
||||
void zmq::stream_engine_t::timer_event (int id_)
|
||||
{
|
||||
zmq_assert (id_ == handshake_timer_id);
|
||||
has_handshake_timer = false;
|
||||
|
||||
// handshake timer expired before handshake completed, so engine fails
|
||||
error ();
|
||||
}
|
||||
|
@ -68,6 +68,7 @@ namespace zmq
|
||||
// i_poll_events interface implementation.
|
||||
void in_event ();
|
||||
void out_event ();
|
||||
void timer_event (int id_);
|
||||
|
||||
private:
|
||||
|
||||
@ -114,6 +115,8 @@ namespace zmq
|
||||
size_t add_property (unsigned char *ptr,
|
||||
const char *name, const void *value, size_t value_len);
|
||||
|
||||
void set_handshake_timer();
|
||||
|
||||
// Underlying socket.
|
||||
fd_t s;
|
||||
|
||||
@ -187,6 +190,12 @@ namespace zmq
|
||||
// True iff the engine doesn't have any message to encode.
|
||||
bool output_stopped;
|
||||
|
||||
// ID of the handshake timer
|
||||
enum {handshake_timer_id = 0x40};
|
||||
|
||||
// True is linger timer is running.
|
||||
bool has_handshake_timer;
|
||||
|
||||
// Socket
|
||||
zmq::socket_base_t *socket;
|
||||
|
||||
|
@ -25,6 +25,7 @@ noinst_PROGRAMS = test_system \
|
||||
test_stream \
|
||||
test_stream_empty \
|
||||
test_stream_disconnect \
|
||||
test_stream_timeout \
|
||||
test_disconnect_inproc \
|
||||
test_ctx_options \
|
||||
test_ctx_destroy \
|
||||
|
216
tests/test_stream_timeout.cpp
Normal file
216
tests/test_stream_timeout.cpp
Normal file
@ -0,0 +1,216 @@
|
||||
/*
|
||||
Copyright (c) 2007-2014 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"
|
||||
|
||||
// Read one event off the monitor socket; return value and address
|
||||
// by reference, if not null, and event number by value. Returns -1
|
||||
// in case of error.
|
||||
|
||||
static int
|
||||
get_monitor_event (void *monitor, int *value, char **address)
|
||||
{
|
||||
// First frame in message contains event number and value
|
||||
zmq_msg_t msg;
|
||||
zmq_msg_init (&msg);
|
||||
if (zmq_msg_recv (&msg, monitor, 0) == -1)
|
||||
return -1; // Interruped, presumably
|
||||
assert (zmq_msg_more (&msg));
|
||||
|
||||
uint8_t *data = (uint8_t *) zmq_msg_data (&msg);
|
||||
uint16_t event = *(uint16_t *) (data);
|
||||
if (value)
|
||||
*value = *(uint32_t *) (data + 2);
|
||||
|
||||
// Second frame in message contains event address
|
||||
zmq_msg_init (&msg);
|
||||
if (zmq_msg_recv (&msg, monitor, 0) == -1)
|
||||
return -1; // Interruped, presumably
|
||||
assert (!zmq_msg_more (&msg));
|
||||
|
||||
if (address) {
|
||||
uint8_t *data = (uint8_t *) zmq_msg_data (&msg);
|
||||
size_t size = zmq_msg_size (&msg);
|
||||
*address = (char *) malloc (size + 1);
|
||||
memcpy (*address, data, size);
|
||||
*address [size] = 0;
|
||||
}
|
||||
return event;
|
||||
}
|
||||
|
||||
static void
|
||||
test_stream_handshake_timeout_accept (void)
|
||||
{
|
||||
int rc;
|
||||
|
||||
// Set up our context and sockets
|
||||
void *ctx = zmq_ctx_new ();
|
||||
assert (ctx);
|
||||
|
||||
// We use this socket in raw mode, to make a connection and send nothing
|
||||
void *stream = zmq_socket (ctx, ZMQ_STREAM);
|
||||
assert (stream);
|
||||
|
||||
int zero = 0;
|
||||
rc = zmq_setsockopt (stream, ZMQ_LINGER, &zero, sizeof (zero));
|
||||
assert (rc == 0);
|
||||
rc = zmq_connect (stream, "tcp://localhost:5557");
|
||||
assert (rc == 0);
|
||||
|
||||
// We'll be using this socket to test TCP stream handshake timeout
|
||||
void *dealer = zmq_socket (ctx, ZMQ_DEALER);
|
||||
assert (dealer);
|
||||
rc = zmq_setsockopt (dealer, ZMQ_LINGER, &zero, sizeof (zero));
|
||||
assert (rc == 0);
|
||||
int val, tenth = 100;
|
||||
size_t vsize = sizeof(val);
|
||||
|
||||
// check for the expected default handshake timeout value - 30 sec
|
||||
rc = zmq_getsockopt (dealer, ZMQ_HANDSHAKE_IVL, &val, &vsize);
|
||||
assert (rc == 0);
|
||||
assert (vsize == sizeof(val));
|
||||
assert (val == 30000);
|
||||
// make handshake timeout faster - 1/10 sec
|
||||
rc = zmq_setsockopt (dealer, ZMQ_HANDSHAKE_IVL, &tenth, sizeof (tenth));
|
||||
assert (rc == 0);
|
||||
vsize = sizeof(val);
|
||||
// make sure zmq_setsockopt changed the value
|
||||
rc = zmq_getsockopt (dealer, ZMQ_HANDSHAKE_IVL, &val, &vsize);
|
||||
assert (rc == 0);
|
||||
assert (vsize == sizeof(val));
|
||||
assert (val == tenth);
|
||||
|
||||
// Create and connect a socket for collecting monitor events on dealer
|
||||
void *dealer_mon = zmq_socket (ctx, ZMQ_PAIR);
|
||||
assert (dealer_mon);
|
||||
|
||||
rc = zmq_socket_monitor (dealer, "inproc://monitor-dealer",
|
||||
ZMQ_EVENT_CONNECTED | ZMQ_EVENT_DISCONNECTED | ZMQ_EVENT_ACCEPTED);
|
||||
assert (rc == 0);
|
||||
|
||||
// Connect to the inproc endpoint so we'll get events
|
||||
rc = zmq_connect (dealer_mon, "inproc://monitor-dealer");
|
||||
assert (rc == 0);
|
||||
|
||||
// bind dealer socket to accept connection from non-sending stream socket
|
||||
rc = zmq_bind (dealer, "tcp://127.0.0.1:5557");
|
||||
assert (rc == 0);
|
||||
|
||||
// we should get ZMQ_EVENT_ACCEPTED and then ZMQ_EVENT_DISCONNECTED
|
||||
int event = get_monitor_event (dealer_mon, NULL, NULL);
|
||||
assert (event == ZMQ_EVENT_ACCEPTED);
|
||||
event = get_monitor_event (dealer_mon, NULL, NULL);
|
||||
assert (event == ZMQ_EVENT_DISCONNECTED);
|
||||
|
||||
rc = zmq_close (dealer);
|
||||
assert (rc == 0);
|
||||
|
||||
rc = zmq_close (dealer_mon);
|
||||
assert (rc == 0);
|
||||
|
||||
rc = zmq_close (stream);
|
||||
assert (rc == 0);
|
||||
|
||||
rc = zmq_ctx_term (ctx);
|
||||
assert (rc == 0);
|
||||
}
|
||||
|
||||
static void
|
||||
test_stream_handshake_timeout_connect (void)
|
||||
{
|
||||
int rc;
|
||||
|
||||
// Set up our context and sockets
|
||||
void *ctx = zmq_ctx_new ();
|
||||
assert (ctx);
|
||||
|
||||
// We use this socket in raw mode, to accept a connection and send nothing
|
||||
void *stream = zmq_socket (ctx, ZMQ_STREAM);
|
||||
assert (stream);
|
||||
|
||||
int zero = 0;
|
||||
rc = zmq_setsockopt (stream, ZMQ_LINGER, &zero, sizeof (zero));
|
||||
assert (rc == 0);
|
||||
rc = zmq_bind (stream, "tcp://127.0.0.1:5556");
|
||||
assert (rc == 0);
|
||||
|
||||
// We'll be using this socket to test TCP stream handshake timeout
|
||||
void *dealer = zmq_socket (ctx, ZMQ_DEALER);
|
||||
assert (dealer);
|
||||
rc = zmq_setsockopt (dealer, ZMQ_LINGER, &zero, sizeof (zero));
|
||||
assert (rc == 0);
|
||||
int val, tenth = 100;
|
||||
size_t vsize = sizeof(val);
|
||||
|
||||
// check for the expected default handshake timeout value - 30 sec
|
||||
rc = zmq_getsockopt (dealer, ZMQ_HANDSHAKE_IVL, &val, &vsize);
|
||||
assert (rc == 0);
|
||||
assert (vsize == sizeof(val));
|
||||
assert (val == 30000);
|
||||
// make handshake timeout faster - 1/10 sec
|
||||
rc = zmq_setsockopt (dealer, ZMQ_HANDSHAKE_IVL, &tenth, sizeof (tenth));
|
||||
assert (rc == 0);
|
||||
vsize = sizeof(val);
|
||||
// make sure zmq_setsockopt changed the value
|
||||
rc = zmq_getsockopt (dealer, ZMQ_HANDSHAKE_IVL, &val, &vsize);
|
||||
assert (rc == 0);
|
||||
assert (vsize == sizeof(val));
|
||||
assert (val == tenth);
|
||||
|
||||
// Create and connect a socket for collecting monitor events on dealer
|
||||
void *dealer_mon = zmq_socket (ctx, ZMQ_PAIR);
|
||||
assert (dealer_mon);
|
||||
|
||||
rc = zmq_socket_monitor (dealer, "inproc://monitor-dealer",
|
||||
ZMQ_EVENT_CONNECTED | ZMQ_EVENT_DISCONNECTED | ZMQ_EVENT_ACCEPTED);
|
||||
assert (rc == 0);
|
||||
|
||||
// Connect to the inproc endpoint so we'll get events
|
||||
rc = zmq_connect (dealer_mon, "inproc://monitor-dealer");
|
||||
assert (rc == 0);
|
||||
|
||||
// connect dealer socket to non-sending stream socket
|
||||
rc = zmq_connect (dealer, "tcp://localhost:5556");
|
||||
assert (rc == 0);
|
||||
|
||||
// we should get ZMQ_EVENT_CONNECTED and then ZMQ_EVENT_DISCONNECTED
|
||||
int event = get_monitor_event (dealer_mon, NULL, NULL);
|
||||
assert (event == ZMQ_EVENT_CONNECTED);
|
||||
event = get_monitor_event (dealer_mon, NULL, NULL);
|
||||
assert (event == ZMQ_EVENT_DISCONNECTED);
|
||||
|
||||
rc = zmq_close (dealer);
|
||||
assert (rc == 0);
|
||||
|
||||
rc = zmq_close (dealer_mon);
|
||||
assert (rc == 0);
|
||||
|
||||
rc = zmq_close (stream);
|
||||
assert (rc == 0);
|
||||
|
||||
rc = zmq_ctx_term (ctx);
|
||||
assert (rc == 0);
|
||||
}
|
||||
|
||||
int main (void)
|
||||
{
|
||||
setup_test_environment();
|
||||
test_stream_handshake_timeout_accept ();
|
||||
test_stream_handshake_timeout_connect ();
|
||||
}
|
Loading…
Reference in New Issue
Block a user