generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-02-28 18:35:41 +01:00
Code Issues Projects Releases Wiki Activity

add option to stop trying to reconnect on ECONNREFUSED (#3831)

Browse Source 
* add option to stop trying to reconnect on ECONNREFUSED
This commit is contained in:
Bill Torpey 2020-02-23 12:17:22 -05:00 committed by GitHub
parent 66ee3ee46c
commit debbe08fb8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 377 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
Makefile.am Normal file → Executable file
View File

@ -1039,7 +1039,8 @@ test_apps += tests/test_poller \
tests/test_app_meta \
tests/test_xpub_manual_last_value \
tests/test_router_notify \
tests/test_peer
tests/test_peer \
tests/test_reconnect_options
tests_test_poller_SOURCES = tests/test_poller.cpp
tests_test_poller_LDADD = ${TESTUTIL_LIBS} src/libzmq.la
@ -1084,6 +1085,10 @@ tests_test_router_notify_CPPFLAGS = ${TESTUTIL_CPPFLAGS}
tests_test_peer_SOURCES = tests/test_peer.cpp
tests_test_peer_LDADD = ${TESTUTIL_LIBS} src/libzmq.la
tests_test_peer_CPPFLAGS = ${TESTUTIL_CPPFLAGS}
tests_test_reconnect_options_SOURCES = tests/test_reconnect_options.cpp
tests_test_reconnect_options_LDADD = ${TESTUTIL_LIBS} src/libzmq.la
tests_test_reconnect_options_CPPFLAGS = ${TESTUTIL_CPPFLAGS}
endif
if ENABLE_STATIC

22
doc/zmq_getsockopt.txt Normal file → Executable file
View File

@ -297,8 +297,8 @@ Applicable socket types:: all but ZMQ_STREAM, only for connection-oriented trans
ZMQ_IDENTITY: Retrieve socket identity
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
This option name is now deprecated. Use ZMQ_ROUTING_ID instead.
ZMQ_IDENTITY remains as an alias for now.
This option name is now deprecated. Use ZMQ_ROUTING_ID instead.
ZMQ_IDENTITY remains as an alias for now.
ZMQ_IMMEDIATE: Retrieve attach-on-connect value
@ -626,6 +626,22 @@ Default value:: 0 (only use ZMQ_RECONNECT_IVL)
Applicable socket types:: all, only for connection-oriented transport
ZMQ_RECONNECT_STOP: Retrieve condition where reconnection will stop
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_RECONNECT_STOP' option shall retrieve the conditions under which
automatic reconnection will stop.
The 'ZMQ_RECONNECT_STOP_CONN_REFUSED' option will stop reconnection when 0MQ
receives the ECONNREFUSED return code from the connect. This indicates that
there is no code bound to the specified endpoint.
[horizontal]
Option value type:: int
Option value unit:: 'ZMQ_RECONNECT_STOP_CONN_REFUSED'
Default value:: 0
Applicable socket types:: all, only for connection-oriented transports
ZMQ_RECOVERY_IVL: Get multicast recovery interval
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_RECOVERY_IVL' option shall retrieve the recovery interval for
@ -823,7 +839,7 @@ ZMQ_ZAP_DOMAIN: Retrieve RFC 27 authentication domain
The 'ZMQ_ZAP_DOMAIN' option shall retrieve the last ZAP domain set for
the socket. The returned value shall be a NULL-terminated string and MAY
be empty. An empty string means that ZAP authentication is disabled.
be empty. An empty string means that ZAP authentication is disabled.
The returned size SHALL include the terminating null byte.
[horizontal]

61
doc/zmq_setsockopt.txt Normal file → Executable file
View File

@ -87,30 +87,30 @@ Default value:: not set
Applicable socket types:: all, when using TCP or UDP transports.
ZMQ_CONNECT_RID: Assign the next outbound connection id
ZMQ_CONNECT_RID: Assign the next outbound connection id
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
This option name is now deprecated. Use ZMQ_CONNECT_ROUTING_ID instead.
ZMQ_CONNECT_RID remains as an alias for now.
This option name is now deprecated. Use ZMQ_CONNECT_ROUTING_ID instead.
ZMQ_CONNECT_RID remains as an alias for now.
ZMQ_CONNECT_ROUTING_ID: Assign the next outbound routing id
ZMQ_CONNECT_ROUTING_ID: Assign the next outbound routing id
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_CONNECT_ROUTING_ID' option sets the peer id of the peer connected
via the next zmq_connect() call, such that that connection is immediately ready for
The 'ZMQ_CONNECT_ROUTING_ID' option sets the peer id of the peer connected
via the next zmq_connect() call, such that that connection is immediately ready for
data transfer with the given routing id. This option applies only to the first
subsequent call to zmq_connect(), zmq_connect() calls thereafter use the default
connection behaviour.
subsequent call to zmq_connect(), zmq_connect() calls thereafter use the default
connection behaviour.
Typical use is to set this socket option ahead of each zmq_connect() call.
Each connection MUST be assigned a unique routing id. Assigning a
routing id that is already in use is not allowed.
Typical use is to set this socket option ahead of each zmq_connect() call.
Each connection MUST be assigned a unique routing id. Assigning a
routing id that is already in use is not allowed.
Useful when connecting ROUTER to ROUTER, or STREAM to STREAM, as it
allows for immediate sending to peers. Outbound routing id framing requirements
Useful when connecting ROUTER to ROUTER, or STREAM to STREAM, as it
allows for immediate sending to peers. Outbound routing id framing requirements
for ROUTER and STREAM sockets apply.
The routing id must be from 1 to 255 bytes long and MAY NOT start with
a zero byte (such routing ids are reserved for internal use by the 0MQ
The routing id must be from 1 to 255 bytes long and MAY NOT start with
a zero byte (such routing ids are reserved for internal use by the 0MQ
infrastructure).
[horizontal]
@ -371,8 +371,8 @@ Applicable socket types:: all, when using connection-oriented transports
ZMQ_IDENTITY: Set socket identity
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
This option name is now deprecated. Use ZMQ_ROUTING_ID instead.
ZMQ_IDENTITY remains as an alias for now.
This option name is now deprecated. Use ZMQ_ROUTING_ID instead.
ZMQ_IDENTITY remains as an alias for now.
ZMQ_IMMEDIATE: Queue messages only to completed connections
@ -701,6 +701,23 @@ Default value:: 0 (only use ZMQ_RECONNECT_IVL)
Applicable socket types:: all, only for connection-oriented transports
ZMQ_RECONNECT_STOP: Set condition where reconnection will stop
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_RECONNECT_STOP' option shall set the conditions under which automatic
reconnection will stop. This can be useful when a process binds to a
wild-card port, where the OS supplies an ephemeral port.
The 'ZMQ_RECONNECT_STOP_CONN_REFUSED' option will stop reconnection when 0MQ
receives the ECONNREFUSED return code from the connect. This indicates that
there is no code bound to the specified endpoint.
[horizontal]
Option value type:: int
Option value unit:: ZMQ_RECONNECT_STOP_CONN_REFUSED
Default value:: 0
Applicable socket types:: all, only for connection-oriented transports
ZMQ_RECOVERY_IVL: Set multicast recovery interval
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_RECOVERY_IVL' option shall set the recovery interval for multicast
@ -813,7 +830,7 @@ Applicable socket types:: ZMQ_ROUTER
ZMQ_ROUTING_ID: Set socket routing id
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_ROUTING_ID' option shall set the routing id of the specified 'socket'
when connecting to a ROUTER socket.
when connecting to a ROUTER socket.
A routing id must be at least one byte and at most 255 bytes long. Identities
starting with a zero byte are reserved for use by the 0MQ infrastructure.
@ -1094,7 +1111,7 @@ ZMQ_XPUB_MANUAL: change the subscription handling to manual
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Sets the 'XPUB' socket subscription handling mode manual/automatic.
A value of '0' is the default and subscription requests will be handled automatically.
A value of '1' will change the subscription requests handling to manual,
A value of '1' will change the subscription requests handling to manual,
with manual mode subscription requests are not added to the subscription list.
To add subscription the user need to call setsockopt with ZMQ_SUBSCRIBE on XPUB socket.
@ -1125,7 +1142,7 @@ Applicable socket types:: ZMQ_XPUB
ZMQ_XPUB_NODROP: do not silently drop messages if SENDHWM is reached
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Sets the 'XPUB' socket behaviour to return error EAGAIN if SENDHWM is
reached and the message could not be send.
reached and the message could not be send.
A value of `0` is the default and drops the message silently when the peers
SNDHWM is reached. A value of `1` returns an 'EAGAIN' error code if the
@ -1172,8 +1189,8 @@ Applicable socket types:: ZMQ_XSUB, ZMQ_XPUB
ZMQ_ZAP_DOMAIN: Set RFC 27 authentication domain
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Sets the domain for ZAP (ZMQ RFC 27) authentication. A ZAP domain must be
specified to enable authentication. When the ZAP domain is empty, which is
Sets the domain for ZAP (ZMQ RFC 27) authentication. A ZAP domain must be
specified to enable authentication. When the ZAP domain is empty, which is
the default, ZAP authentication is disabled. This is not compatible with
previous versions of libzmq, so it can be controlled by ZMQ_ZAP_ENFORCE_DOMAIN
which for now is disabled by default.

3
include/zmq.h Normal file → Executable file
View File

@ -677,7 +677,10 @@ ZMQ_EXPORT void zmq_threadclose (void *thread_);
#define ZMQ_WSS_HOSTNAME 106
#define ZMQ_WSS_TRUST_SYSTEM 107
#define ZMQ_ONLY_FIRST_SUBSCRIBE 108
#define ZMQ_RECONNECT_STOP 109
/* DRAFT ZMQ_RECONNECT_STOP options */
#define ZMQ_RECONNECT_STOP_CONN_REFUSED 0x1
/* DRAFT Context options */
#define ZMQ_ZERO_COPY_RECV 10

1
src/command.hpp
View File

@ -74,6 +74,7 @@ __declspec(align (64))
reap,
reaped,
inproc_connected,
conn_failed,
pipe_peer_stats,
pipe_stats_publish,
done

17
src/object.cpp
View File

@ -161,6 +161,10 @@ void zmq::object_t::process_command (const command_t &cmd_)
process_seqnum ();
break;
case command_t::conn_failed:
process_conn_failed ();
break;
case command_t::done:
default:
zmq_assert (false);
@ -257,6 +261,14 @@ void zmq::object_t::send_attach (session_base_t *destination_,
send_command (cmd);
}
void zmq::object_t::send_conn_failed (session_base_t *destination_)
{
command_t cmd;
cmd.destination = destination_;
cmd.type = command_t::conn_failed;
send_command (cmd);
}
void zmq::object_t::send_bind (own_t *destination_,
pipe_t *pipe_,
bool inc_seqnum_)
@ -528,6 +540,11 @@ void zmq::object_t::process_seqnum ()
zmq_assert (false);
}
void zmq::object_t::process_conn_failed ()
{
zmq_assert (false);
}
void zmq::object_t::send_command (const command_t &cmd_)
{
_ctx->send_command (cmd_.destination->get_tid (), cmd_);

4
src/object.hpp
View File

@ -117,6 +117,8 @@ class object_t
void send_reap (zmq::socket_base_t *socket_);
void send_reaped ();
void send_done ();
void send_conn_failed (zmq::session_base_t *destination_);
// These handlers can be overridden by the derived objects. They are
// called when command arrives from another thread.
@ -144,6 +146,8 @@ class object_t
virtual void process_term_endpoint (std::string *endpoint_);
virtual void process_reap (zmq::socket_base_t *socket_);
virtual void process_reaped ();
virtual void process_conn_failed ();
// Special handler called after a command that requires a seqnum
// was processed. The implementation should catch up with its counter

15
src/options.cpp
View File

@ -210,6 +210,7 @@ zmq::options_t::options_t () :
linger (-1),
connect_timeout (0),
tcp_maxrt (0),
reconnect_stop (0),
reconnect_ivl (100),
reconnect_ivl_max (0),
backlog (100),
@ -393,6 +394,13 @@ int zmq::options_t::setsockopt (int option_,
}
break;
case ZMQ_RECONNECT_STOP:
if (is_int) {
reconnect_stop = value;
return 0;
}
break;
case ZMQ_RECONNECT_IVL:
if (is_int && value >= -1) {
reconnect_ivl = value;
@ -933,6 +941,13 @@ int zmq::options_t::getsockopt (int option_,
}
break;
case ZMQ_RECONNECT_STOP:
if (is_int) {
*value = reconnect_stop;
return 0;
}
break;
case ZMQ_RECONNECT_IVL:
if (is_int) {
*value = reconnect_ivl;

4
src/options.hpp
View File

@ -116,6 +116,10 @@ struct options_t
// Default 0 (unused)
int tcp_maxrt;
// Disable reconnect under certain conditions
// Default 0
int reconnect_stop;
// Minimum interval between attempts to reconnect, in milliseconds.
// Default 100ms
int reconnect_ivl;

7
src/session_base.cpp
View File

@ -523,6 +523,13 @@ void zmq::session_base_t::timer_event (int id_)
_pipe->terminate (false);
}
void zmq::session_base_t::process_conn_failed ()
{
std::string *ep = new (std::string);
_addr->to_string (*ep);
send_term_endpoint (_socket, ep);
}
void zmq::session_base_t::reconnect ()
{
// For delayed connect situations, terminate the pipe

1
src/session_base.hpp
View File

@ -141,6 +141,7 @@ class session_base_t : public own_t, public io_object_t, public i_pipe_events
void process_plug () ZMQ_FINAL;
void process_attach (zmq::i_engine *engine_) ZMQ_FINAL;
void process_term (int linger_) ZMQ_FINAL;
void process_conn_failed ();
// i_poll_events handlers.
void timer_event (int id_) ZMQ_FINAL;

4
src/stream_connecter_base.cpp Normal file → Executable file
View File

@ -55,10 +55,10 @@ zmq::stream_connecter_base_t::stream_connecter_base_t (
_s (retired_fd),
_handle (static_cast<handle_t> (NULL)),
_socket (session_->get_socket ()),
_session (session_),
_delayed_start (delayed_start_),
_reconnect_timer_started (false),
_current_reconnect_ivl (options.reconnect_ivl)
_current_reconnect_ivl (options.reconnect_ivl),
_session (session_)
{
zmq_assert (_addr);
_addr->to_string (_endpoint);

7
src/stream_connecter_base.hpp Normal file → Executable file
View File

@ -91,9 +91,6 @@ class stream_connecter_base_t : public own_t, public io_object_t
// Socket
zmq::socket_base_t *const _socket;
// Reference to the session we belong to.
zmq::session_base_t *const _session;
private:
// ID of the timer used to delay the reconnection.
enum
@ -118,6 +115,10 @@ class stream_connecter_base_t : public own_t, public io_object_t
int _current_reconnect_ivl;
ZMQ_NON_COPYABLE_NOR_MOVABLE (stream_connecter_base_t)
protected:
// Reference to the session we belong to.
zmq::session_base_t *const _session;
};
}

10
src/tcp_connecter.cpp Normal file → Executable file
View File

@ -103,6 +103,15 @@ void zmq::tcp_connecter_t::out_event ()
const fd_t fd = connect ();
if (fd == retired_fd
&& ((options.reconnect_stop & ZMQ_RECONNECT_STOP_CONN_REFUSED)
&& errno == ECONNREFUSED)) {
send_conn_failed (_session);
close ();
terminate ();
return;
}
// Handle the error condition by attempt to reconnect.
if (fd == retired_fd || !tune_socket (fd)) {
close ();
@ -266,6 +275,7 @@ zmq::fd_t zmq::tcp_connecter_t::connect ()
|| err == WSAENOBUFS) {
wsa_assert_no (err);
}
errno = wsa_error_to_errno (err);
return retired_fd;
}
#else

3
src/zmq_draft.h
View File

@ -64,7 +64,10 @@
#define ZMQ_WSS_HOSTNAME 106
#define ZMQ_WSS_TRUST_SYSTEM 107
#define ZMQ_ONLY_FIRST_SUBSCRIBE 108
#define ZMQ_RECONNECT_STOP 109
/* DRAFT ZMQ_RECONNECT_STOP options */
#define ZMQ_RECONNECT_STOP_CONN_REFUSED 0x1
/* DRAFT Context options */
#define ZMQ_ZERO_COPY_RECV 10

4
tests/CMakeLists.txt Normal file → Executable file
View File

@ -163,6 +163,7 @@ if(ENABLE_DRAFTS)
test_router_notify
test_xpub_manual_last_value
test_peer
test_reconnect_options
)
endif()
@ -248,7 +249,7 @@ foreach(test ${tests})
link_directories(${test} PRIVATE "${ZeroMQ_SOURCE_DIR}/../lib")
endif()
endif()
if(RT_LIBRARY)
target_link_libraries(${test} ${RT_LIBRARY})
endif()
@ -292,7 +293,6 @@ if(ZMQ_HAVE_CURVE AND NOT ZMQ_USE_TWEETNACL)
endif()
set_tests_properties(test_security_zap PROPERTIES TIMEOUT 60)
set_tests_properties(test_reconnect_ivl PROPERTIES TIMEOUT 15)
#Check whether all tests in the current folder are present

240
tests/test_reconnect_options.cpp Executable file
View File

@ -0,0 +1,240 @@
/*
Copyright (c) 2017 Contributors as noted in the AUTHORS file
This file is part of libzmq, the ZeroMQ core engine in C++.
libzmq is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser General Public License (LGPL) as published
by the Free Software Foundation; either version 3 of the License, or
(at your option) any later version.
As a special exception, the Contributors give you permission to link
this library with independent modules to produce an executable,
regardless of the license terms of these independent modules, and to
copy and distribute the resulting executable under terms of your choice,
provided that you also meet, for each linked independent module, the
terms and conditions of the license of that module. An independent
module is a module which is not derived from or based on this library.
If you modify this library, you must extend this exception to your
version of the library.
libzmq 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 <assert.h>
#include "testutil.hpp"
#include "testutil_unity.hpp"
#include "testutil_monitoring.hpp"
#include <unity.h>
// test behavior with (mostly) default values
void reconnect_default ()
{
// setup pub socket
void *pub = test_context_socket (ZMQ_PUB);
// Bind pub socket
TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (pub, ENDPOINT_0));
// setup sub socket
void *sub = test_context_socket (ZMQ_SUB);
// Monitor all events on sub
TEST_ASSERT_SUCCESS_ERRNO (
zmq_socket_monitor (sub, "inproc://monitor-sub", ZMQ_EVENT_ALL));
// Create socket for collecting monitor events
void *sub_mon = test_context_socket (ZMQ_PAIR);
// Connect so they'll get events
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub_mon, "inproc://monitor-sub"));
// set reconnect interval so only a single reconnect is tried
int interval = 60 * 1000;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (sub, ZMQ_RECONNECT_IVL, &interval, sizeof (interval)));
// connect to pub
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub, ENDPOINT_0));
// confirm that we get following events
expect_monitor_event (sub_mon, ZMQ_EVENT_CONNECT_DELAYED);
expect_monitor_event (sub_mon, ZMQ_EVENT_CONNECTED);
expect_monitor_event (sub_mon, ZMQ_EVENT_HANDSHAKE_SUCCEEDED);
// close the pub socket
test_context_socket_close_zero_linger (pub);
// confirm that we get following events
expect_monitor_event (sub_mon, ZMQ_EVENT_DISCONNECTED);
expect_monitor_event (sub_mon, ZMQ_EVENT_CONNECT_RETRIED);
// ZMQ_EVENT_CONNECT_RETRIED should be last event, because of timeout set above
int event;
char *event_address;
int rc = get_monitor_event_with_timeout (sub_mon, &event, &event_address,
2 * 1000);
assert (rc == -1);
// Close sub
// TODO why does this use zero_linger?
test_context_socket_close_zero_linger (sub);
// Close monitor
// TODO why does this use zero_linger?
test_context_socket_close_zero_linger (sub_mon);
}
// test successful reconnect
void reconnect_success ()
{
// setup pub socket
void *pub = test_context_socket (ZMQ_PUB);
// Bind pub socket
TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (pub, ENDPOINT_0));
// setup sub socket
void *sub = test_context_socket (ZMQ_SUB);
// Monitor all events on sub
TEST_ASSERT_SUCCESS_ERRNO (
zmq_socket_monitor (sub, "inproc://monitor-sub", ZMQ_EVENT_ALL));
// Create socket for collecting monitor events
void *sub_mon = test_context_socket (ZMQ_PAIR);
// Connect so they'll get events
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub_mon, "inproc://monitor-sub"));
// set reconnect interval so only a single reconnect is tried
int interval = 1 * 1000;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (sub, ZMQ_RECONNECT_IVL, &interval, sizeof (interval)));
// connect to pub
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub, ENDPOINT_0));
// confirm that we get following events
expect_monitor_event (sub_mon, ZMQ_EVENT_CONNECT_DELAYED);
expect_monitor_event (sub_mon, ZMQ_EVENT_CONNECTED);
expect_monitor_event (sub_mon, ZMQ_EVENT_HANDSHAKE_SUCCEEDED);
// close the pub socket
test_context_socket_close_zero_linger (pub);
// confirm that we get following events
expect_monitor_event (sub_mon, ZMQ_EVENT_DISCONNECTED);
expect_monitor_event (sub_mon, ZMQ_EVENT_CONNECT_RETRIED);
// ZMQ_EVENT_CONNECT_RETRIED should be last event, because of timeout set above
int event;
char *event_address;
int rc = get_monitor_event_with_timeout (sub_mon, &event, &event_address,
SETTLE_TIME);
assert (rc == -1);
// Now re-bind pub socket and wait for re-connect
pub = test_context_socket (ZMQ_PUB);
TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (pub, ENDPOINT_0));
msleep (SETTLE_TIME);
// confirm that we get following events
expect_monitor_event (sub_mon, ZMQ_EVENT_CONNECT_DELAYED);
expect_monitor_event (sub_mon, ZMQ_EVENT_CONNECTED);
expect_monitor_event (sub_mon, ZMQ_EVENT_HANDSHAKE_SUCCEEDED);
// ZMQ_EVENT_HANDSHAKE_SUCCEEDED should be last event
rc = get_monitor_event_with_timeout (sub_mon, &event, &event_address,
SETTLE_TIME);
assert (rc == -1);
// Close sub
// TODO why does this use zero_linger?
test_context_socket_close_zero_linger (sub);
test_context_socket_close_zero_linger (pub);
// Close monitor
// TODO why does this use zero_linger?
test_context_socket_close_zero_linger (sub_mon);
}
// test stopping reconnect on connection refused
void reconnect_stop_on_refused ()
{
// setup pub socket
void *pub = test_context_socket (ZMQ_PUB);
// Bind pub socket
TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (pub, ENDPOINT_0));
// setup sub socket
void *sub = test_context_socket (ZMQ_SUB);
// Monitor all events on sub
TEST_ASSERT_SUCCESS_ERRNO (
zmq_socket_monitor (sub, "inproc://monitor-sub", ZMQ_EVENT_ALL));
// Create socket for collecting monitor events
void *sub_mon = test_context_socket (ZMQ_PAIR);
// Connect so they'll get events
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub_mon, "inproc://monitor-sub"));
// set option to stop reconnecting on error
int stopReconnectOnError = ZMQ_RECONNECT_STOP_CONN_REFUSED;
TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (sub, ZMQ_RECONNECT_STOP,
&stopReconnectOnError,
sizeof (stopReconnectOnError)));
// connect to pub
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub, ENDPOINT_0));
// confirm that we get following events
expect_monitor_event (sub_mon, ZMQ_EVENT_CONNECT_DELAYED);
expect_monitor_event (sub_mon, ZMQ_EVENT_CONNECTED);
expect_monitor_event (sub_mon, ZMQ_EVENT_HANDSHAKE_SUCCEEDED);
// close the pub socket
test_context_socket_close_zero_linger (pub);
// confirm that we get following events
expect_monitor_event (sub_mon, ZMQ_EVENT_DISCONNECTED);
expect_monitor_event (sub_mon, ZMQ_EVENT_CONNECT_RETRIED);
expect_monitor_event (sub_mon, ZMQ_EVENT_CONNECT_DELAYED);
expect_monitor_event (sub_mon, ZMQ_EVENT_CLOSED);
// ZMQ_EVENT_CLOSED should be last event, because of ZMQ_RECONNECT_STOP set above
int event = 0;
char *event_address;
int rc = get_monitor_event_with_timeout (sub_mon, &event, &event_address,
2 * 1000);
int limit = 0;
while ((rc != -1) && (++limit < 1000)) {
print_unexpected_event_stderr(event, rc, 0, -1);
rc = get_monitor_event_with_timeout (sub_mon, &event, &event_address,
2 * 1000);
}
// Close sub
// TODO why does this use zero_linger?
test_context_socket_close_zero_linger (sub);
// Close monitor
// TODO why does this use zero_linger?
test_context_socket_close_zero_linger (sub_mon);
}
void setUp ()
{
setup_test_context ();
}
void tearDown ()
{
teardown_test_context ();
}
int main (void)
{
setup_test_environment ();
UNITY_BEGIN ();
RUN_TEST (reconnect_default);
RUN_TEST (reconnect_success);
RUN_TEST (reconnect_stop_on_refused);
return UNITY_END ();
}