generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-01-31 14:39:55 +01:00
Code Issues Projects Releases Wiki Activity

Merge pull request #3168 from bluca/recv_sub_cancel

Browse Source 
Problem: can't process ZMTP 3.1 cancel/subscribe commands
This commit is contained in:
Simon Giesecke 2018-06-25 14:00:42 +02:00 committed by GitHub
commit 0d66067411
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 473 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
.gitignore vendored
View File

@ -1,12 +1,13 @@
syntax: glob # for hg-git users
/Makefile
builds/Makefile
builds/msvc/Makefile
builds/deprecated-msvc/Makefile
configure
doc/Makefile
Makefile.in
configure
libtool
libunity.a
config
config.status
config.log
@ -141,8 +142,21 @@ test_bind_after_connect_tcp
test_sodium
test_zmq_poll_fd
test_reconnect_ivl
test_address_tipc
test_app_meta
test_security_zap
test_socket_null
test_xpub_verbose
test_mock_pub_sub
unittest_ip_resolver
unittest_mtrie
unittest_poller
unittest_udp_address
unittest_ypipe
tests/test*.log
tests/test*.trs
unittests/unittest*.log
unittests/unittest*.trs
src/platform.hpp*
src/stamp-h1
local_lat

16
Makefile.am
View File

@ -295,6 +295,12 @@ src_libzmq_la_LDFLAGS = \
@LIBZMQ_EXTRA_LDFLAGS@ \
-Wl,--version-script=$(srcdir)/src/libzmq.vers
else
if ON_GNU
src_libzmq_la_LDFLAGS = \
-version-info @LTVER@ \
@LIBZMQ_EXTRA_LDFLAGS@ \
-Wl,--version-script=$(srcdir)/src/libzmq.vers
else
src_libzmq_la_LDFLAGS = \
-version-info @LTVER@ \
@LIBZMQ_EXTRA_LDFLAGS@ \
@ -303,6 +309,7 @@ endif
endif
endif
endif
endif
src_libzmq_la_CPPFLAGS = $(CODE_COVERAGE_CPPFLAGS) $(LIBUNWIND_CFLAGS)
src_libzmq_la_CFLAGS = $(CODE_COVERAGE_CFLAGS) $(LIBUNWIND_CFLAGS)
@ -438,6 +445,7 @@ test_apps = \
tests/test_bind_after_connect_tcp \
tests/test_sodium \
tests/test_reconnect_ivl \
tests/test_mock_pub_sub \
tests/test_socket_null
UNITY_CPPFLAGS = -I$(top_srcdir)/external/unity -DUNITY_USE_COMMAND_LINE_ARGS -DUNITY_EXCLUDE_FLOAT
@ -692,6 +700,10 @@ tests_test_reconnect_ivl_SOURCES = tests/test_reconnect_ivl.cpp
tests_test_reconnect_ivl_LDADD = src/libzmq.la ${UNITY_LIBS}
tests_test_reconnect_ivl_CPPFLAGS = ${UNITY_CPPFLAGS}
tests_test_mock_pub_sub_SOURCES = tests/test_mock_pub_sub.cpp
tests_test_mock_pub_sub_LDADD = src/libzmq.la ${UNITY_LIBS}
tests_test_mock_pub_sub_CPPFLAGS = ${UNITY_CPPFLAGS}
if HAVE_CURVE
test_apps += \
@ -856,12 +868,12 @@ if HAVE_VMCI
test_apps += test_pair_vmci test_reqrep_vmci
test_pair_vmci_SOURCES = tests/test_pair_vmci.cpp
test_pair_vmci_LDADD = libzmq.la
test_pair_vmci_LDADD = src/libzmq.la
test_pair_vmci_LDFLAGS = @LIBZMQ_VMCI_LDFLAGS@
test_pair_vmci_CXXFLAGS = @LIBZMQ_VMCI_CXXFLAGS@
test_reqrep_vmci_SOURCES = tests/test_reqrep_vmci.cpp
test_reqrep_vmci_LDADD = libzmq.la
test_reqrep_vmci_LDADD = src/libzmq.la
test_reqrep_vmci_LDFLAGS = @LIBZMQ_VMCI_LDFLAGS@
test_reqrep_vmci_CXXFLAGS = @LIBZMQ_VMCI_CXXFLAGS@

6
configure.ac
View File

@ -207,6 +207,12 @@ case "${host_os}" in
;;
*freebsd*)
# Define on FreeBSD to enable all library features
case "${host_os}" in
# On Debian/kFreeBSD with gnu set the --version-script flag
kfreebsd*-gnu*)
libzmq_on_gnu="yes"
;;
esac
CPPFLAGS="-D__BSD_VISIBLE $CPPFLAGS"
AC_DEFINE(ZMQ_HAVE_FREEBSD, 1, [Have FreeBSD OS])
;;

37
src/msg.cpp
View File

@ -444,6 +444,43 @@ bool zmq::msg_t::is_leave () const
return _u.base.type == type_leave;
}
bool zmq::msg_t::is_ping () const
{
return (_u.base.flags & CMD_TYPE_MASK) == ping;
}
bool zmq::msg_t::is_pong () const
{
return (_u.base.flags & CMD_TYPE_MASK) == pong;
}
size_t zmq::msg_t::command_body_size () const
{
if (this->is_ping () || this->is_pong ())
return this->size () - ping_cmd_name_size;
if (this->is_subscribe ())
return this->size () - sub_cmd_name_size;
if (this->is_cancel ())
return this->size () - cancel_cmd_name_size;
return 0;
}
void *zmq::msg_t::command_body ()
{
unsigned char *data = NULL;
if (this->is_ping () || this->is_pong ())
data =
static_cast<unsigned char *> (this->data ()) + ping_cmd_name_size;
if (this->is_subscribe ())
data = static_cast<unsigned char *> (this->data ()) + sub_cmd_name_size;
if (this->is_cancel ())
data =
static_cast<unsigned char *> (this->data ()) + cancel_cmd_name_size;
return data;
}
void zmq::msg_t::add_refs (int refs_)
{
zmq_assert (refs_ >= 0);

31
src/msg.hpp
View File

@ -39,6 +39,9 @@
#include "atomic_counter.hpp"
#include "metadata.hpp"
// bits 2-5
#define CMD_TYPE_MASK 0x1c
// Signature for free function to deallocate the message content.
// Note that it has to be declared as "C" so that it is the same as
// zmq_free_fn defined in zmq.h.
@ -75,6 +78,12 @@ class msg_t
{
more = 1, // Followed by more parts
command = 2, // Command frame (see ZMTP spec)
// Command types, use only bits 2-5 and compare with ==, not bitwise,
// a command can never be of more that one type at the same time
ping = 4,
pong = 8,
subscribe = 12,
cancel = 16,
credential = 32,
routing_id = 64,
shared = 128
@ -115,6 +124,22 @@ class msg_t
bool is_delimiter () const;
bool is_join () const;
bool is_leave () const;
bool is_ping () const;
bool is_pong () const;
// These are called on each message received by the session_base class,
// so get them inlined to avoid the overhead of 2 function calls per msg
inline bool is_subscribe () const
{
return (_u.base.flags & CMD_TYPE_MASK) == subscribe;
}
inline bool is_cancel () const
{
return (_u.base.flags & CMD_TYPE_MASK) == cancel;
}
size_t command_body_size () const;
void *command_body ();
bool is_vsm () const;
bool is_cmsg () const;
bool is_lmsg () const;
@ -145,6 +170,12 @@ class msg_t
max_vsm_size =
msg_t_size - (sizeof (metadata_t *) + 3 + 16 + sizeof (uint32_t))
};
enum
{
ping_cmd_name_size = 5, // 4PING
cancel_cmd_name_size = 7, // 6CANCEL
sub_cmd_name_size = 10 // 9SUBSCRIBE
};
private:
zmq::atomic_counter_t *refcnt ();

4
src/pgm_sender.cpp
View File

@ -51,6 +51,10 @@ zmq::pgm_sender_t::pgm_sender_t (io_thread_t *parent_,
more_flag (false),
pgm_socket (false, options_),
options (options_),
handle (static_cast<handle_t> (NULL)),
uplink_handle (static_cast<handle_t> (NULL)),
rdata_notify_handle (static_cast<handle_t> (NULL)),
pending_notify_handle (static_cast<handle_t> (NULL)),
out_buffer (NULL),
out_buffer_size (0),
write_size (0)

4
src/session_base.cpp
View File

@ -163,7 +163,9 @@ int zmq::session_base_t::pull_msg (msg_t *msg_)
int zmq::session_base_t::push_msg (msg_t *msg_)
{
if (msg_->flags () & msg_t::command)
// pass subscribe/cancel to the sockets
if ((msg_->flags () & msg_t::command) && !msg_->is_subscribe ()
&& !msg_->is_cancel ())
return 0;
if (_pipe && _pipe->write (msg_)) {
int rc = msg_->init ();

61
src/stream_engine.cpp
View File

@ -1038,18 +1038,20 @@ void zmq::stream_engine_t::timer_event (int id_)
int zmq::stream_engine_t::produce_ping_message (msg_t *msg_)
{
int rc = 0;
// 16-bit TTL + \4PING == 7
const size_t ping_ttl_len = msg_t::ping_cmd_name_size + 2;
zmq_assert (_mechanism != NULL);
// 16-bit TTL + \4PING == 7
rc = msg_->init_size (7);
rc = msg_->init_size (ping_ttl_len);
errno_assert (rc == 0);
msg_->set_flags (msg_t::command);
// Copy in the command message
memcpy (msg_->data (), "\4PING", 5);
memcpy (msg_->data (), "\4PING", msg_t::ping_cmd_name_size);
uint16_t ttl_val = htons (_options.heartbeat_ttl);
memcpy ((static_cast<uint8_t *> (msg_->data ())) + 5, &ttl_val,
sizeof (ttl_val));
memcpy ((static_cast<uint8_t *> (msg_->data ()))
+ msg_t::ping_cmd_name_size,
&ttl_val, sizeof (ttl_val));
rc = _mechanism->encode (msg_);
_next_msg = &stream_engine_t::pull_and_encode;
@ -1075,11 +1077,17 @@ int zmq::stream_engine_t::produce_pong_message (msg_t *msg_)
int zmq::stream_engine_t::process_heartbeat_message (msg_t *msg_)
{
if (memcmp (msg_->data (), "\4PING", 5) == 0) {
if (msg_->is_ping ()) {
// 16-bit TTL + \4PING == 7
const size_t ping_ttl_len = msg_t::ping_cmd_name_size + 2;
const size_t ping_max_ctx_len = 16;
uint16_t remote_heartbeat_ttl;
// Get the remote heartbeat TTL to setup the timer
memcpy (&remote_heartbeat_ttl,
static_cast<uint8_t *> (msg_->data ()) + 5, 2);
static_cast<uint8_t *> (msg_->data ())
+ msg_t::ping_cmd_name_size,
ping_ttl_len - msg_t::ping_cmd_name_size);
remote_heartbeat_ttl = ntohs (remote_heartbeat_ttl);
// The remote heartbeat is in 10ths of a second
// so we multiply it by 100 to get the timer interval in ms.
@ -1095,14 +1103,18 @@ int zmq::stream_engine_t::process_heartbeat_message (msg_t *msg_)
// here and store it. Truncate it if it's too long.
// Given the engine goes straight to out_event, sequential PINGs will
// not be a problem.
size_t context_len = msg_->size () - 7 > 16 ? 16 : msg_->size () - 7;
int rc = _pong_msg.init_size (5 + context_len);
size_t context_len = msg_->size () - ping_ttl_len > ping_max_ctx_len
? ping_max_ctx_len
: msg_->size () - ping_ttl_len;
int rc = _pong_msg.init_size (msg_t::ping_cmd_name_size + context_len);
errno_assert (rc == 0);
_pong_msg.set_flags (msg_t::command);
memcpy (_pong_msg.data (), "\4PONG", 5);
memcpy (_pong_msg.data (), "\4PONG", msg_t::ping_cmd_name_size);
if (context_len > 0)
memcpy ((static_cast<uint8_t *> (_pong_msg.data ())) + 5,
(static_cast<uint8_t *> (msg_->data ())) + 7, context_len);
memcpy ((static_cast<uint8_t *> (_pong_msg.data ()))
+ msg_t::ping_cmd_name_size,
(static_cast<uint8_t *> (msg_->data ())) + ping_ttl_len,
context_len);
_next_msg = &stream_engine_t::produce_pong_message;
out_event ();
@ -1115,15 +1127,28 @@ int zmq::stream_engine_t::process_command_message (msg_t *msg_)
{
const uint8_t cmd_name_size =
*(static_cast<const uint8_t *> (msg_->data ()));
const size_t ping_name_size = msg_t::ping_cmd_name_size - 1;
const size_t sub_name_size = msg_t::sub_cmd_name_size - 1;
const size_t cancel_name_size = msg_t::cancel_cmd_name_size - 1;
// Malformed command
if (msg_->size () < cmd_name_size + sizeof (cmd_name_size))
if (unlikely (msg_->size () < cmd_name_size + sizeof (cmd_name_size)))
return -1;
const uint8_t *cmd_name =
(static_cast<const uint8_t *> (msg_->data ())) + 1;
if (cmd_name_size == 4
&& (memcmp (cmd_name, "PING", cmd_name_size) == 0
|| memcmp (cmd_name, "PONG", cmd_name_size) == 0))
uint8_t *cmd_name = (static_cast<uint8_t *> (msg_->data ())) + 1;
if (cmd_name_size == ping_name_size
&& memcmp (cmd_name, "PING", cmd_name_size) == 0)
msg_->set_flags (zmq::msg_t::ping);
if (cmd_name_size == ping_name_size
&& memcmp (cmd_name, "PONG", cmd_name_size) == 0)
msg_->set_flags (zmq::msg_t::pong);
if (cmd_name_size == sub_name_size
&& memcmp (cmd_name, "SUBSCRIBE", cmd_name_size) == 0)
msg_->set_flags (zmq::msg_t::subscribe);
if (cmd_name_size == cancel_name_size
&& memcmp (cmd_name, "CANCEL", cmd_name_size) == 0)
msg_->set_flags (zmq::msg_t::cancel);
if (msg_->is_ping () || msg_->is_pong ())
return process_heartbeat_message (msg_);
return 0;

118
src/xpub.cpp
View File

@ -88,57 +88,89 @@ void zmq::xpub_t::xread_activated (pipe_t *pipe_)
// There are some subscriptions waiting. Let's process them.
msg_t sub;
while (pipe_->read (&sub)) {
// Apply the subscription to the trie
const unsigned char *const data =
static_cast<const unsigned char *> (sub.data ());
const size_t size = sub.size ();
metadata_t *metadata = sub.metadata ();
if (size > 0 && (*data == 0 || *data == 1)) {
if (_manual) {
// Store manual subscription to use on termination
if (*data == 0)
_manual_subscriptions.rm (data + 1, size - 1, pipe_);
else
_manual_subscriptions.add (data + 1, size - 1, pipe_);
unsigned char *msg_data = static_cast<unsigned char *> (sub.data ()),
*data = NULL;
size_t size = 0;
bool subscribe = false;
// Apply the subscription to the trie
if (sub.is_subscribe () || sub.is_cancel ()) {
data = static_cast<unsigned char *> (sub.command_body ());
size = sub.command_body_size ();
subscribe = sub.is_subscribe ();
} else if (sub.size () > 0) {
unsigned char first = *msg_data;
if (first == 0 || first == 1) {
data = msg_data + 1;
size = sub.size () - 1;
subscribe = first == 1;
}
} else {
// Process user message coming upstream from xsub socket
_pending_data.push_back (blob_t (msg_data, sub.size ()));
if (metadata)
metadata->add_ref ();
_pending_metadata.push_back (metadata);
_pending_flags.push_back (sub.flags ());
sub.close ();
continue;
}
if (_manual) {
// Store manual subscription to use on termination
if (!subscribe)
_manual_subscriptions.rm (data, size, pipe_);
else
_manual_subscriptions.add (data, size, pipe_);
_pending_pipes.push_back (pipe_);
// ZMTP 3.1 hack: we need to support sub/cancel commands, but
// we can't give them back to userspace as it would be an API
// breakage since the payload of the message is completely
// different. Manually craft an old-style message instead.
data = data - 1;
size = size + 1;
if (subscribe)
*data = 1;
else
*data = 0;
_pending_data.push_back (blob_t (data, size));
if (metadata)
metadata->add_ref ();
_pending_metadata.push_back (metadata);
_pending_flags.push_back (0);
} else {
bool notify;
if (!subscribe) {
mtrie_t::rm_result rm_result =
_subscriptions.rm (data, size, pipe_);
// TODO reconsider what to do if rm_result == mtrie_t::not_found
notify = rm_result != mtrie_t::values_remain || _verbose_unsubs;
} else {
bool first_added = _subscriptions.add (data, size, pipe_);
notify = first_added || _verbose_subs;
}
// If the request was a new subscription, or the subscription
// was removed, or verbose mode is enabled, store it so that
// it can be passed to the user on next recv call.
if (options.type == ZMQ_XPUB && notify) {
data = data - 1;
size = size + 1;
if (subscribe)
*data = 1;
else
*data = 0;
_pending_pipes.push_back (pipe_);
_pending_data.push_back (blob_t (data, size));
if (metadata)
metadata->add_ref ();
_pending_metadata.push_back (metadata);
_pending_flags.push_back (0);
} else {
bool notify;
if (*data == 0) {
mtrie_t::rm_result rm_result =
_subscriptions.rm (data + 1, size - 1, pipe_);
// TODO reconsider what to do if rm_result == mtrie_t::not_found
notify =
rm_result != mtrie_t::values_remain || _verbose_unsubs;
} else {
bool first_added =
_subscriptions.add (data + 1, size - 1, pipe_);
notify = first_added || _verbose_subs;
}
// If the request was a new subscription, or the subscription
// was removed, or verbose mode is enabled, store it so that
// it can be passed to the user on next recv call.
if (options.type == ZMQ_XPUB && notify) {
_pending_data.push_back (blob_t (data, size));
if (metadata)
metadata->add_ref ();
_pending_metadata.push_back (metadata);
_pending_flags.push_back (0);
}
}
} else {
// Process user message coming upstream from xsub socket
_pending_data.push_back (blob_t (data, size));
if (metadata)
metadata->add_ref ();
_pending_metadata.push_back (metadata);
_pending_flags.push_back (sub.flags ());
}
sub.close ();
}

22
src/xsub.cpp
View File

@ -96,18 +96,32 @@ int zmq::xsub_t::xsend (msg_t *msg_)
size_t size = msg_->size ();
unsigned char *data = static_cast<unsigned char *> (msg_->data ());
if (size > 0 && *data == 1) {
if (msg_->is_subscribe () || (size > 0 && *data == 1)) {
// Process subscribe message
// This used to filter out duplicate subscriptions,
// however this is alread done on the XPUB side and
// doing it here as well breaks ZMQ_XPUB_VERBOSE
// when there are forwarding devices involved.
_subscriptions.add (data + 1, size - 1);
if (msg_->is_subscribe ()) {
data = static_cast<unsigned char *> (msg_->command_body ());
size = msg_->command_body_size ();
} else {
data = data + 1;
size = size - 1;
}
_subscriptions.add (data, size);
return _dist.send_to_all (msg_);
}
if (size > 0 && *data == 0) {
if (msg_->is_cancel () || (size > 0 && *data == 0)) {
// Process unsubscribe message
if (_subscriptions.rm (data + 1, size - 1))
if (msg_->is_cancel ()) {
data = static_cast<unsigned char *> (msg_->command_body ());
size = msg_->command_body_size ();
} else {
data = data + 1;
size = size - 1;
}
if (_subscriptions.rm (data, size))
return _dist.send_to_all (msg_);
} else
// User message sent upstream to XPUB socket

1
tests/CMakeLists.txt
View File

@ -72,6 +72,7 @@ set(tests
test_monitor
test_socket_null
test_reconnect_ivl
test_mock_pub_sub
)
if(ZMQ_HAVE_CURVE)
list(APPEND tests

226
tests/test_mock_pub_sub.cpp Normal file
View File

@ -0,0 +1,226 @@
/*
Copyright (c) 2018 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"
#include "testutil_unity.hpp"
#if defined(ZMQ_HAVE_WINDOWS)
#include <winsock2.h>
#include <ws2tcpip.h>
#include <stdexcept>
#define close closesocket
typedef SOCKET raw_socket;
#else
#include <arpa/inet.h>
typedef int raw_socket;
#endif
void setUp ()
{
setup_test_context ();
}
void tearDown ()
{
teardown_test_context ();
}
// 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_)
{
for (int i = 0; i < 2; i++) {
// First frame in message contains event number and value
zmq_msg_t msg;
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_init (&msg));
if (zmq_msg_recv (&msg, monitor_, ZMQ_DONTWAIT) == -1) {
msleep (SETTLE_TIME);
continue; // Interrupted, presumably
}
TEST_ASSERT_TRUE (zmq_msg_more (&msg));
uint8_t *data = (uint8_t *) zmq_msg_data (&msg);
uint16_t event = *(uint16_t *) (data);
// Second frame in message contains event address
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_init (&msg));
if (zmq_msg_recv (&msg, monitor_, 0) == -1) {
return -1; // Interrupted, presumably
}
TEST_ASSERT_FALSE (zmq_msg_more (&msg));
return event;
}
return -1;
}
static void recv_with_retry (raw_socket fd_, char *buffer_, int bytes_)
{
int received = 0;
while (true) {
int rc = TEST_ASSERT_SUCCESS_RAW_ERRNO (
recv (fd_, buffer_ + received, bytes_ - received, 0));
TEST_ASSERT_GREATER_THAN_INT (0, rc);
received += rc;
TEST_ASSERT_LESS_OR_EQUAL_INT (bytes_, received);
if (received == bytes_)
break;
}
}
static void mock_handshake (raw_socket fd_)
{
const uint8_t zmtp_greeting[33] = {0xff, 0, 0, 0, 0, 0, 0, 0, 0,
0x7f, 3, 0, 'N', 'U', 'L', 'L', 0};
char buffer[128];
memset (buffer, 0, sizeof (buffer));
memcpy (buffer, zmtp_greeting, sizeof (zmtp_greeting));
int rc = TEST_ASSERT_SUCCESS_RAW_ERRNO (send (fd_, buffer, 64, 0));
TEST_ASSERT_EQUAL_INT (64, rc);
recv_with_retry (fd_, buffer, 64);
const uint8_t zmtp_ready[27] = {
4, 25, 5, 'R', 'E', 'A', 'D', 'Y', 11, 'S', 'o', 'c', 'k', 'e',
't', '-', 'T', 'y', 'p', 'e', 0, 0, 0, 3, 'S', 'U', 'B'};
memset (buffer, 0, sizeof (buffer));
memcpy (buffer, zmtp_ready, 27);
rc = TEST_ASSERT_SUCCESS_RAW_ERRNO (send (fd_, buffer, 27, 0));
TEST_ASSERT_EQUAL_INT (27, rc);
// greeting - XPUB so one extra byte
recv_with_retry (fd_, buffer, 28);
}
static void prep_server_socket (void **server_out_,
void **mon_out_,
char *endpoint_,
size_t ep_length_)
{
// We'll be using this socket in raw mode
void *server = test_context_socket (ZMQ_XPUB);
int value = 0;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (server, ZMQ_LINGER, &value, sizeof (value)));
bind_loopback_ipv4 (server, endpoint_, ep_length_);
// Create and connect a socket for collecting monitor events on xpub
void *server_mon = test_context_socket (ZMQ_PAIR);
TEST_ASSERT_SUCCESS_ERRNO (zmq_socket_monitor (
server, "inproc://monitor-dealer",
ZMQ_EVENT_CONNECTED | ZMQ_EVENT_DISCONNECTED | ZMQ_EVENT_ACCEPTED));
// Connect to the inproc endpoint so we'll get events
TEST_ASSERT_SUCCESS_ERRNO (
zmq_connect (server_mon, "inproc://monitor-dealer"));
*server_out_ = server;
*mon_out_ = server_mon;
}
static void test_mock_sub (bool sub_command)
{
int rc;
char my_endpoint[MAX_SOCKET_STRING];
void *server, *server_mon;
prep_server_socket (&server, &server_mon, my_endpoint, MAX_SOCKET_STRING);
struct sockaddr_in ip4addr;
raw_socket s;
ip4addr.sin_family = AF_INET;
ip4addr.sin_port = htons (atoi (strrchr (my_endpoint, ':') + 1));
#if defined(ZMQ_HAVE_WINDOWS) && (_WIN32_WINNT < 0x0600)
ip4addr.sin_addr.s_addr = inet_addr ("127.0.0.1");
#else
inet_pton (AF_INET, "127.0.0.1", &ip4addr.sin_addr);
#endif
s = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
rc = TEST_ASSERT_SUCCESS_RAW_ERRNO (
connect (s, (struct sockaddr *) &ip4addr, sizeof ip4addr));
TEST_ASSERT_GREATER_THAN_INT (-1, rc);
// Mock a ZMTP 3 client so we can forcibly try sub commands
mock_handshake (s);
// By now everything should report as connected
rc = get_monitor_event (server_mon);
TEST_ASSERT_EQUAL_INT (ZMQ_EVENT_ACCEPTED, rc);
if (sub_command) {
const uint8_t sub[13] = {4, 11, 9, 'S', 'U', 'B', 'S',
'C', 'R', 'I', 'B', 'E', 'A'};
rc =
TEST_ASSERT_SUCCESS_RAW_ERRNO (send (s, (const char *) sub, 13, 0));
TEST_ASSERT_EQUAL_INT (13, rc);
} else {
const uint8_t sub[4] = {0, 2, 1, 'A'};
rc = TEST_ASSERT_SUCCESS_RAW_ERRNO (send (s, (const char *) sub, 4, 0));
TEST_ASSERT_EQUAL_INT (4, rc);
}
char buffer[16];
memset (buffer, 0, sizeof (buffer));
rc = zmq_recv (server, buffer, 2, 0);
TEST_ASSERT_EQUAL_INT (2, rc);
TEST_ASSERT_EQUAL_INT (0, memcmp (buffer, "\1A", 2));
rc = zmq_send (server, "ALOL", 4, 0);
TEST_ASSERT_EQUAL_INT (4, rc);
memset (buffer, 0, sizeof (buffer));
recv_with_retry (s, buffer, 6);
TEST_ASSERT_EQUAL_INT (0, memcmp (buffer, "\0\4ALOL", 6));
close (s);
test_context_socket_close (server);
test_context_socket_close (server_mon);
}
void test_mock_sub_command ()
{
test_mock_sub (true);
}
void test_mock_sub_legacy ()
{
test_mock_sub (false);
}
int main (void)
{
setup_test_environment ();
UNITY_BEGIN ();
RUN_TEST (test_mock_sub_command);
RUN_TEST (test_mock_sub_legacy);
return UNITY_END ();
}