mirror of
https://github.com/zeromq/libzmq.git
synced 2024-12-25 22:30:51 +08:00
da31917f4f
Relicense permission collected from all relevant authors as tallied at: https://github.com/rlenferink/libzmq-relicense/blob/master/checklist.md The relicense grants are collected under RELICENSE/ and will be moved to the above repository in a later commit. Fixes https://github.com/zeromq/libzmq/issues/2376
111 lines
3.7 KiB
C++
111 lines
3.7 KiB
C++
/* SPDX-License-Identifier: MPL-2.0 */
|
|
|
|
#include "testutil.hpp"
|
|
#include "testutil_unity.hpp"
|
|
|
|
#include <string.h>
|
|
|
|
SETUP_TEARDOWN_TESTCONTEXT
|
|
|
|
/// Initialize a zeromq message with a given null-terminated string
|
|
#define ZMQ_PREPARE_STRING(msg, data, size) \
|
|
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_init (&msg)); \
|
|
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_init_size (&msg, size + 1)); \
|
|
memcpy (zmq_msg_data (&msg), data, size + 1);
|
|
|
|
static int publicationsReceived = 0;
|
|
static bool isSubscribed = false;
|
|
|
|
void test_disconnect_inproc ()
|
|
{
|
|
void *pub_socket = test_context_socket (ZMQ_XPUB);
|
|
void *sub_socket = test_context_socket (ZMQ_SUB);
|
|
TEST_ASSERT_SUCCESS_ERRNO (
|
|
zmq_setsockopt (sub_socket, ZMQ_SUBSCRIBE, "foo", 3));
|
|
|
|
TEST_ASSERT_SUCCESS_ERRNO (
|
|
zmq_bind (pub_socket, "inproc://someInProcDescriptor"));
|
|
|
|
int more;
|
|
size_t more_size = sizeof (more);
|
|
|
|
for (int iteration = 0;; ++iteration) {
|
|
zmq_pollitem_t items[] = {
|
|
{sub_socket, 0, ZMQ_POLLIN, 0}, // read publications
|
|
{pub_socket, 0, ZMQ_POLLIN, 0}, // read subscriptions
|
|
};
|
|
int rc = zmq_poll (items, 2, 100);
|
|
|
|
if (items[1].revents & ZMQ_POLLIN) {
|
|
for (more = 1; more;) {
|
|
zmq_msg_t msg;
|
|
zmq_msg_init (&msg);
|
|
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&msg, pub_socket, 0));
|
|
const char *const buffer =
|
|
static_cast<const char *> (zmq_msg_data (&msg));
|
|
|
|
if (buffer[0] == 0) {
|
|
TEST_ASSERT_TRUE (isSubscribed);
|
|
isSubscribed = false;
|
|
} else {
|
|
TEST_ASSERT_FALSE (isSubscribed);
|
|
isSubscribed = true;
|
|
}
|
|
|
|
TEST_ASSERT_SUCCESS_ERRNO (
|
|
zmq_getsockopt (pub_socket, ZMQ_RCVMORE, &more, &more_size));
|
|
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_close (&msg));
|
|
}
|
|
}
|
|
|
|
if (items[0].revents & ZMQ_POLLIN) {
|
|
more = 1;
|
|
for (more = 1; more;) {
|
|
zmq_msg_t msg;
|
|
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_init (&msg));
|
|
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&msg, sub_socket, 0));
|
|
TEST_ASSERT_SUCCESS_ERRNO (
|
|
zmq_getsockopt (sub_socket, ZMQ_RCVMORE, &more, &more_size));
|
|
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_close (&msg));
|
|
}
|
|
publicationsReceived++;
|
|
}
|
|
if (iteration == 1) {
|
|
TEST_ASSERT_SUCCESS_ERRNO (
|
|
zmq_connect (sub_socket, "inproc://someInProcDescriptor"));
|
|
msleep (SETTLE_TIME);
|
|
}
|
|
if (iteration == 4) {
|
|
TEST_ASSERT_SUCCESS_ERRNO (
|
|
zmq_disconnect (sub_socket, "inproc://someInProcDescriptor"));
|
|
}
|
|
if (iteration > 4 && rc == 0)
|
|
break;
|
|
|
|
zmq_msg_t channel_envlp;
|
|
ZMQ_PREPARE_STRING (channel_envlp, "foo", 3);
|
|
TEST_ASSERT_SUCCESS_ERRNO (
|
|
zmq_msg_send (&channel_envlp, pub_socket, ZMQ_SNDMORE));
|
|
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_close (&channel_envlp));
|
|
|
|
zmq_msg_t message;
|
|
ZMQ_PREPARE_STRING (message, "this is foo!", 12);
|
|
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_send (&message, pub_socket, 0));
|
|
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_close (&message));
|
|
}
|
|
TEST_ASSERT_EQUAL_INT (3, publicationsReceived);
|
|
TEST_ASSERT_FALSE (isSubscribed);
|
|
|
|
test_context_socket_close (pub_socket);
|
|
test_context_socket_close (sub_socket);
|
|
}
|
|
|
|
int main (int, char **)
|
|
{
|
|
setup_test_environment ();
|
|
|
|
UNITY_BEGIN ();
|
|
RUN_TEST (test_disconnect_inproc);
|
|
return UNITY_END ();
|
|
}
|