From 2f98f7034bb2fb7c59cbb9363c90ca9b2a2a0da1 Mon Sep 17 00:00:00 2001 From: imkcy9 Date: Sat, 18 May 2019 05:12:32 +0800 Subject: [PATCH] Support XPub socket send last value caching to last subscription pipe with ZMQ_XPUB_MANUAL_LAST_VALUE. (#3511) * Add ZMQ_XPUB_MANUAL_LAST_VALUE * Surpport xpub send last value caching to one pipe with ZMQ_XPUB_MANUAL_LAST_VALUE * Add test_xpubub_manual_last_value * Add relicense and doc --- RELICENSE/imkcy9.md | 16 + doc/zmq_setsockopt.txt | 17 + include/zmq.h | 1 + src/xpub.cpp | 30 ++ src/xpub.hpp | 8 + src/zmq_draft.h | 1 + tests/CMakeLists.txt | 1 + tests/test_xpub_manual_last_value.cpp | 510 ++++++++++++++++++++++++++ 8 files changed, 584 insertions(+) create mode 100644 RELICENSE/imkcy9.md create mode 100644 tests/test_xpub_manual_last_value.cpp diff --git a/RELICENSE/imkcy9.md b/RELICENSE/imkcy9.md new file mode 100644 index 00000000..66235c28 --- /dev/null +++ b/RELICENSE/imkcy9.md @@ -0,0 +1,16 @@ +# Permission to Relicense under MPLv2 or any other OSI approved license chosen by the current ZeroMQ BDFL + +This is a statement by Chengye Ke +that grants permission to relicense its copyrights in the libzmq C++ +library (ZeroMQ) under the Mozilla Public License v2 (MPLv2) or any other +Open Source Initiative approved license chosen by the current ZeroMQ +BDFL (Benevolent Dictator for Life). + +A portion of the commits made by the Github handle "imkcy9", with +commit author "Chengye Ke " or +"Chengye Ke ", are copyright of Chengye Ke. +This document hereby grants the libzmq project team to relicense libzmq, +including all past, present and future contributions of the author listed above. + +Chengye Ke +2019/05/18 \ No newline at end of file diff --git a/doc/zmq_setsockopt.txt b/doc/zmq_setsockopt.txt index 89dc7b81..3f1b8c8c 100644 --- a/doc/zmq_setsockopt.txt +++ b/doc/zmq_setsockopt.txt @@ -1073,6 +1073,23 @@ Default value:: 0 Applicable socket types:: ZMQ_XPUB +ZMQ_XPUB_MANUAL_LAST_VALUE: change the subscription handling to manual +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +This option is similar to ZMQ_XPUB_MANUAL. +What is the difference, ZMQ_XPUB_MANUAL_LAST_VALUE sets the 'XPUB' socket +behaviour to send the first message to the last subscriber after the 'XPUB' socket +recieve a subscription and call setsockopt with ZMQ_SUBSCRIBE on 'XPUB' socket. +This prevent duplicated message when use last value caching(LVC). + +NOTE: in DRAFT state, not yet available in stable releases. + +[horizontal] +Option value type:: int +Option value unit:: 0, 1 +Default value:: 0 +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 diff --git a/include/zmq.h b/include/zmq.h index b614855a..8dd2790c 100644 --- a/include/zmq.h +++ b/include/zmq.h @@ -655,6 +655,7 @@ ZMQ_EXPORT void zmq_threadclose (void *thread_); #define ZMQ_METADATA 95 #define ZMQ_MULTICAST_LOOP 96 #define ZMQ_ROUTER_NOTIFY 97 +#define ZMQ_XPUB_MANUAL_LAST_VALUE 98 /* DRAFT Context options */ #define ZMQ_ZERO_COPY_RECV 10 diff --git a/src/xpub.cpp b/src/xpub.cpp index 2652c8c0..689d286a 100644 --- a/src/xpub.cpp +++ b/src/xpub.cpp @@ -44,6 +44,9 @@ zmq::xpub_t::xpub_t (class ctx_t *parent_, uint32_t tid_, int sid_) : _more (false), _lossy (true), _manual (false), +#ifdef ZMQ_BUILD_DRAFT_API + _send_last_pipe (false), +#endif _pending_pipes (), _welcome_msg () { @@ -190,6 +193,9 @@ int zmq::xpub_t::xsetsockopt (int option_, size_t optvallen_) { if (option_ == ZMQ_XPUB_VERBOSE || option_ == ZMQ_XPUB_VERBOSER +#ifdef ZMQ_BUILD_DRAFT_API + || option_ == ZMQ_XPUB_MANUAL_LAST_VALUE +#endif || option_ == ZMQ_XPUB_NODROP || option_ == ZMQ_XPUB_MANUAL) { if (optvallen_ != sizeof (int) || *static_cast (optval_) < 0) { @@ -202,6 +208,11 @@ int zmq::xpub_t::xsetsockopt (int option_, } else if (option_ == ZMQ_XPUB_VERBOSER) { _verbose_subs = (*static_cast (optval_) != 0); _verbose_unsubs = _verbose_subs; +#ifdef ZMQ_BUILD_DRAFT_API + } else if (option_ == ZMQ_XPUB_MANUAL_LAST_VALUE) { + _manual = (*static_cast (optval_) != 0); + _send_last_pipe = _manual; +#endif } else if (option_ == ZMQ_XPUB_NODROP) _lossy = (*static_cast (optval_) == 0); else if (option_ == ZMQ_XPUB_MANUAL) @@ -265,14 +276,33 @@ void zmq::xpub_t::mark_as_matching (pipe_t *pipe_, xpub_t *self_) self_->_dist.match (pipe_); } +#ifdef ZMQ_BUILD_DRAFT_API +void zmq::xpub_t::mark_last_pipe_as_matching (pipe_t *pipe_, xpub_t *self_) +{ + if (self_->_last_pipe == pipe_) + self_->_dist.match (pipe_); +} +#endif + int zmq::xpub_t::xsend (msg_t *msg_) { bool msg_more = (msg_->flags () & msg_t::more) != 0; // For the first part of multi-part message, find the matching pipes. if (!_more) { +#ifdef ZMQ_BUILD_DRAFT_API + if (_manual && _last_pipe && _send_last_pipe) { + _subscriptions.match (static_cast (msg_->data ()), + msg_->size (), mark_last_pipe_as_matching, + this); + _last_pipe = NULL; + } else + _subscriptions.match (static_cast (msg_->data ()), + msg_->size (), mark_as_matching, this); +#else _subscriptions.match (static_cast (msg_->data ()), msg_->size (), mark_as_matching, this); +#endif // If inverted matching is used, reverse the selection now if (options.invert_matching) { _dist.reverse_match (); diff --git a/src/xpub.hpp b/src/xpub.hpp index d8032249..21456faf 100644 --- a/src/xpub.hpp +++ b/src/xpub.hpp @@ -99,6 +99,14 @@ class xpub_t : public socket_base_t // Subscriptions will not bed added automatically, only after calling set option with ZMQ_SUBSCRIBE or ZMQ_UNSUBSCRIBE bool _manual; +#ifdef ZMQ_BUILD_DRAFT_API + // Send message to the last pipe, only used if xpub is on manual and after calling set option with ZMQ_SUBSCRIBE + bool _send_last_pipe; + + // Function to be applied to match the last pipe. + static void mark_last_pipe_as_matching (zmq::pipe_t *pipe_, xpub_t *arg_); +#endif + // Last pipe that sent subscription message, only used if xpub is on manual pipe_t *_last_pipe; diff --git a/src/zmq_draft.h b/src/zmq_draft.h index 13f94562..ba3e13ba 100644 --- a/src/zmq_draft.h +++ b/src/zmq_draft.h @@ -52,6 +52,7 @@ #define ZMQ_METADATA 95 #define ZMQ_MULTICAST_LOOP 96 #define ZMQ_ROUTER_NOTIFY 97 +#define ZMQ_XPUB_MANUAL_LAST_VALUE 98 /* DRAFT Context options */ #define ZMQ_ZERO_COPY_RECV 10 diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index ccae3977..b0f15534 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -153,6 +153,7 @@ if(ENABLE_DRAFTS) test_dgram test_app_meta test_router_notify + test_xpub_manual_last_value ) endif() diff --git a/tests/test_xpub_manual_last_value.cpp b/tests/test_xpub_manual_last_value.cpp new file mode 100644 index 00000000..c21be0d1 --- /dev/null +++ b/tests/test_xpub_manual_last_value.cpp @@ -0,0 +1,510 @@ +/* + Copyright (c) 2007-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 . +*/ + +#include "testutil.hpp" +#include "testutil_unity.hpp" + +SETUP_TEARDOWN_TESTCONTEXT + +void test_basic () +{ + // Create a publisher + void *pub = test_context_socket (ZMQ_XPUB); + int manual = 1; + TEST_ASSERT_SUCCESS_ERRNO ( + zmq_setsockopt (pub, ZMQ_XPUB_MANUAL_LAST_VALUE, &manual, 4)); + TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (pub, "inproc://soname")); + + // Create a subscriber + void *sub = test_context_socket (ZMQ_XSUB); + TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub, "inproc://soname")); + + // Subscribe for A + const char subscription[] = {1, 'A', 0}; + send_string_expect_success (sub, subscription, 0); + + // Receive subscriptions from subscriber + recv_string_expect_success (pub, subscription, 0); + + // Subscribe socket for B instead + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (pub, ZMQ_SUBSCRIBE, "B", 1)); + + // Sending A message and B Message + send_string_expect_success (pub, "A", 0); + send_string_expect_success (pub, "B", 0); + + recv_string_expect_success (sub, "B", ZMQ_DONTWAIT); + + // Clean up. + test_context_socket_close (pub); + test_context_socket_close (sub); +} + +void test_unsubscribe_manual () +{ + // Create a publisher + void *pub = test_context_socket (ZMQ_XPUB); + TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (pub, "inproc://soname")); + + // set pub socket options + int manual = 1; + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (pub, ZMQ_XPUB_MANUAL_LAST_VALUE, + &manual, sizeof (manual))); + + // Create a subscriber + void *sub = test_context_socket (ZMQ_XSUB); + TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub, "inproc://soname")); + + // Subscribe for A + const uint8_t subscription1[] = {1, 'A'}; + send_array_expect_success (sub, subscription1, 0); + + // Subscribe for B + const uint8_t subscription2[] = {1, 'B'}; + send_array_expect_success (sub, subscription2, 0); + + char buffer[3]; + + // Receive subscription "A" from subscriber + recv_array_expect_success (pub, subscription1, 0); + + // Subscribe socket for XA instead + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (pub, ZMQ_SUBSCRIBE, "XA", 2)); + + // Receive subscription "B" from subscriber + recv_array_expect_success (pub, subscription2, 0); + + // Subscribe socket for XB instead + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (pub, ZMQ_SUBSCRIBE, "XB", 2)); + + // Unsubscribe from A + const uint8_t unsubscription1[2] = {0, 'A'}; + send_array_expect_success (sub, unsubscription1, 0); + + // Receive unsubscription "A" from subscriber + recv_array_expect_success (pub, unsubscription1, 0); + + // Unsubscribe socket from XA instead + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (pub, ZMQ_UNSUBSCRIBE, "XA", 2)); + + // Sending messages XA, XB + send_string_expect_success (pub, "XA", 0); + send_string_expect_success (pub, "XB", 0); + + // Subscriber should receive XB only + recv_string_expect_success (sub, "XB", ZMQ_DONTWAIT); + + // Close subscriber + test_context_socket_close (sub); + + // Receive unsubscription "B" + const char unsubscription2[2] = {0, 'B'}; + TEST_ASSERT_EQUAL_INT ( + sizeof unsubscription2, + TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (pub, buffer, sizeof buffer, 0))); + TEST_ASSERT_EQUAL_INT8_ARRAY (unsubscription2, buffer, + sizeof unsubscription2); + + // Unsubscribe socket from XB instead + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (pub, ZMQ_UNSUBSCRIBE, "XB", 2)); + + // Clean up. + test_context_socket_close (pub); +} + +void test_xpub_proxy_unsubscribe_on_disconnect () +{ + const uint8_t topic_buff[] = {"1"}; + const uint8_t payload_buff[] = {"X"}; + + char my_endpoint_backend[MAX_SOCKET_STRING]; + char my_endpoint_frontend[MAX_SOCKET_STRING]; + + int manual = 1; + + // proxy frontend + void *xsub_proxy = test_context_socket (ZMQ_XSUB); + bind_loopback_ipv4 (xsub_proxy, my_endpoint_frontend, + sizeof my_endpoint_frontend); + + // proxy backend + void *xpub_proxy = test_context_socket (ZMQ_XPUB); + TEST_ASSERT_SUCCESS_ERRNO ( + zmq_setsockopt (xpub_proxy, ZMQ_XPUB_MANUAL_LAST_VALUE, &manual, 4)); + bind_loopback_ipv4 (xpub_proxy, my_endpoint_backend, + sizeof my_endpoint_backend); + + // publisher + void *pub = test_context_socket (ZMQ_PUB); + TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (pub, my_endpoint_frontend)); + + // first subscriber subscribes + void *sub1 = test_context_socket (ZMQ_SUB); + TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub1, my_endpoint_backend)); + TEST_ASSERT_SUCCESS_ERRNO ( + zmq_setsockopt (sub1, ZMQ_SUBSCRIBE, topic_buff, 1)); + + // wait + msleep (SETTLE_TIME); + + // proxy reroutes and confirms subscriptions + const uint8_t subscription[2] = {1, *topic_buff}; + recv_array_expect_success (xpub_proxy, subscription, ZMQ_DONTWAIT); + TEST_ASSERT_SUCCESS_ERRNO ( + zmq_setsockopt (xpub_proxy, ZMQ_SUBSCRIBE, topic_buff, 1)); + send_array_expect_success (xsub_proxy, subscription, 0); + + // second subscriber subscribes + void *sub2 = test_context_socket (ZMQ_SUB); + TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub2, my_endpoint_backend)); + TEST_ASSERT_SUCCESS_ERRNO ( + zmq_setsockopt (sub2, ZMQ_SUBSCRIBE, topic_buff, 1)); + + // wait + msleep (SETTLE_TIME); + + // proxy reroutes + recv_array_expect_success (xpub_proxy, subscription, ZMQ_DONTWAIT); + TEST_ASSERT_SUCCESS_ERRNO ( + zmq_setsockopt (xpub_proxy, ZMQ_SUBSCRIBE, topic_buff, 1)); + send_array_expect_success (xsub_proxy, subscription, 0); + + // wait + msleep (SETTLE_TIME); + + // let publisher send a msg + send_array_expect_success (pub, topic_buff, ZMQ_SNDMORE); + send_array_expect_success (pub, payload_buff, 0); + + // wait + msleep (SETTLE_TIME); + + // proxy reroutes data messages to subscribers + recv_array_expect_success (xsub_proxy, topic_buff, ZMQ_DONTWAIT); + recv_array_expect_success (xsub_proxy, payload_buff, ZMQ_DONTWAIT); + + // send 2 messages + send_array_expect_success (xpub_proxy, topic_buff, ZMQ_SNDMORE); + send_array_expect_success (xpub_proxy, payload_buff, 0); + send_array_expect_success (xpub_proxy, topic_buff, ZMQ_SNDMORE); + send_array_expect_success (xpub_proxy, payload_buff, 0); + + // wait + msleep (SETTLE_TIME); + + // sub2 will get 2 messages because the last subscription is sub2. + recv_array_expect_success (sub2, topic_buff, ZMQ_DONTWAIT); + recv_array_expect_success (sub2, payload_buff, ZMQ_DONTWAIT); + recv_array_expect_success (sub2, topic_buff, ZMQ_DONTWAIT); + recv_array_expect_success (sub2, payload_buff, ZMQ_DONTWAIT); + + recv_array_expect_success (sub1, topic_buff, ZMQ_DONTWAIT); + recv_array_expect_success (sub1, payload_buff, ZMQ_DONTWAIT); + + // Disconnect both subscribers + test_context_socket_close (sub1); + test_context_socket_close (sub2); + + // wait + msleep (SETTLE_TIME); + + // unsubscribe messages are passed from proxy to publisher + const uint8_t unsubscription[] = {0, *topic_buff}; + recv_array_expect_success (xpub_proxy, unsubscription, 0); + TEST_ASSERT_SUCCESS_ERRNO ( + zmq_setsockopt (xpub_proxy, ZMQ_UNSUBSCRIBE, topic_buff, 1)); + send_array_expect_success (xsub_proxy, unsubscription, 0); + + // should receive another unsubscribe msg + recv_array_expect_success (xpub_proxy, unsubscription, 0); + TEST_ASSERT_SUCCESS_ERRNO ( + zmq_setsockopt (xpub_proxy, ZMQ_UNSUBSCRIBE, topic_buff, 1)); + send_array_expect_success (xsub_proxy, unsubscription, 0); + + // wait + msleep (SETTLE_TIME); + + // let publisher send a msg + send_array_expect_success (pub, topic_buff, ZMQ_SNDMORE); + send_array_expect_success (pub, payload_buff, 0); + + // wait + msleep (SETTLE_TIME); + + // nothing should come to the proxy + char buffer[1]; + TEST_ASSERT_FAILURE_ERRNO ( + EAGAIN, zmq_recv (xsub_proxy, buffer, sizeof buffer, ZMQ_DONTWAIT)); + + test_context_socket_close (pub); + test_context_socket_close (xpub_proxy); + test_context_socket_close (xsub_proxy); +} + +void test_missing_subscriptions () +{ + const char *topic1 = "1"; + const char *topic2 = "2"; + const char *payload = "X"; + + char my_endpoint_backend[MAX_SOCKET_STRING]; + char my_endpoint_frontend[MAX_SOCKET_STRING]; + + int manual = 1; + + // proxy frontend + void *xsub_proxy = test_context_socket (ZMQ_XSUB); + bind_loopback_ipv4 (xsub_proxy, my_endpoint_frontend, + sizeof my_endpoint_frontend); + + // proxy backend + void *xpub_proxy = test_context_socket (ZMQ_XPUB); + TEST_ASSERT_SUCCESS_ERRNO ( + zmq_setsockopt (xpub_proxy, ZMQ_XPUB_MANUAL_LAST_VALUE, &manual, 4)); + bind_loopback_ipv4 (xpub_proxy, my_endpoint_backend, + sizeof my_endpoint_backend); + + // publisher + void *pub = test_context_socket (ZMQ_PUB); + TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (pub, my_endpoint_frontend)); + + // Here's the problem: because subscribers subscribe in quick succession, + // the proxy is unable to confirm the first subscription before receiving + // the second. This causes the first subscription to get lost. + + // first subscriber + void *sub1 = test_context_socket (ZMQ_SUB); + TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub1, my_endpoint_backend)); + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (sub1, ZMQ_SUBSCRIBE, topic1, 1)); + + // second subscriber + void *sub2 = test_context_socket (ZMQ_SUB); + TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub2, my_endpoint_backend)); + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (sub2, ZMQ_SUBSCRIBE, topic2, 1)); + + // wait + msleep (SETTLE_TIME); + + // proxy now reroutes and confirms subscriptions + const uint8_t subscription1[] = {1, static_cast (topic1[0])}; + recv_array_expect_success (xpub_proxy, subscription1, ZMQ_DONTWAIT); + TEST_ASSERT_SUCCESS_ERRNO ( + zmq_setsockopt (xpub_proxy, ZMQ_SUBSCRIBE, topic1, 1)); + send_array_expect_success (xsub_proxy, subscription1, 0); + + const uint8_t subscription2[] = {1, static_cast (topic2[0])}; + recv_array_expect_success (xpub_proxy, subscription2, ZMQ_DONTWAIT); + TEST_ASSERT_SUCCESS_ERRNO ( + zmq_setsockopt (xpub_proxy, ZMQ_SUBSCRIBE, topic2, 1)); + send_array_expect_success (xsub_proxy, subscription2, 0); + + // wait + msleep (SETTLE_TIME); + + // let publisher send 2 msgs, each with its own topic_buff + send_string_expect_success (pub, topic1, ZMQ_SNDMORE); + send_string_expect_success (pub, payload, 0); + send_string_expect_success (pub, topic2, ZMQ_SNDMORE); + send_string_expect_success (pub, payload, 0); + + // wait + msleep (SETTLE_TIME); + + // proxy reroutes data messages to subscribers + recv_string_expect_success (xsub_proxy, topic1, ZMQ_DONTWAIT); + recv_string_expect_success (xsub_proxy, payload, ZMQ_DONTWAIT); + send_string_expect_success (xpub_proxy, topic1, ZMQ_SNDMORE); + send_string_expect_success (xpub_proxy, payload, 0); + + recv_string_expect_success (xsub_proxy, topic2, ZMQ_DONTWAIT); + recv_string_expect_success (xsub_proxy, payload, ZMQ_DONTWAIT); + send_string_expect_success (xpub_proxy, topic2, ZMQ_SNDMORE); + send_string_expect_success (xpub_proxy, payload, 0); + + // wait + msleep (SETTLE_TIME); + + // only sub2 should now get a message + recv_string_expect_success (sub2, topic2, ZMQ_DONTWAIT); + recv_string_expect_success (sub2, payload, ZMQ_DONTWAIT); + + //recv_string_expect_success (sub1, topic1, ZMQ_DONTWAIT); + //recv_string_expect_success (sub1, payload, ZMQ_DONTWAIT); + + // Clean up + test_context_socket_close (sub1); + test_context_socket_close (sub2); + test_context_socket_close (pub); + test_context_socket_close (xpub_proxy); + test_context_socket_close (xsub_proxy); +} + +void test_unsubscribe_cleanup () +{ + char my_endpoint[MAX_SOCKET_STRING]; + + // Create a publisher + void *pub = test_context_socket (ZMQ_XPUB); + int manual = 1; + TEST_ASSERT_SUCCESS_ERRNO ( + zmq_setsockopt (pub, ZMQ_XPUB_MANUAL_LAST_VALUE, &manual, 4)); + bind_loopback_ipv4 (pub, my_endpoint, sizeof my_endpoint); + + // Create a subscriber + void *sub = test_context_socket (ZMQ_XSUB); + TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub, my_endpoint)); + + // Subscribe for A + const uint8_t subscription1[2] = {1, 'A'}; + send_array_expect_success (sub, subscription1, 0); + + + // Receive subscriptions from subscriber + recv_array_expect_success (pub, subscription1, 0); + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (pub, ZMQ_SUBSCRIBE, "XA", 2)); + + // send 2 messages + send_string_expect_success (pub, "XA", 0); + send_string_expect_success (pub, "XB", 0); + + // receive the single message + recv_string_expect_success (sub, "XA", 0); + + // should be nothing left in the queue + char buffer[2]; + TEST_ASSERT_FAILURE_ERRNO ( + EAGAIN, zmq_recv (sub, buffer, sizeof buffer, ZMQ_DONTWAIT)); + + // close the socket + test_context_socket_close (sub); + + // closing the socket will result in an unsubscribe event + const uint8_t unsubscription[2] = {0, 'A'}; + recv_array_expect_success (pub, unsubscription, 0); + + // this doesn't really do anything + // there is no last_pipe set it will just fail silently + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (pub, ZMQ_UNSUBSCRIBE, "XA", 2)); + + // reconnect + sub = test_context_socket (ZMQ_XSUB); + TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub, my_endpoint)); + + // send a subscription for B + const uint8_t subscription2[2] = {1, 'B'}; + send_array_expect_success (sub, subscription2, 0); + + // receive the subscription, overwrite it to XB + recv_array_expect_success (pub, subscription2, 0); + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (pub, ZMQ_SUBSCRIBE, "XB", 2)); + + // send 2 messages + send_string_expect_success (pub, "XA", 0); + send_string_expect_success (pub, "XB", 0); + + // receive the single message + recv_string_expect_success (sub, "XB", 0); + + // should be nothing left in the queue + TEST_ASSERT_FAILURE_ERRNO ( + EAGAIN, zmq_recv (sub, buffer, sizeof buffer, ZMQ_DONTWAIT)); + + // Clean up. + test_context_socket_close (pub); + test_context_socket_close (sub); +} + +void test_manual_last_value () +{ + // Create a publisher + void *pub = test_context_socket (ZMQ_XPUB); + + int hwm = 2000; + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (pub, ZMQ_SNDHWM, &hwm, 4)); + + // set pub socket options + int manual = 1; + TEST_ASSERT_SUCCESS_ERRNO ( + zmq_setsockopt (pub, ZMQ_XPUB_MANUAL_LAST_VALUE, &manual, 4)); + + TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (pub, "inproc://soname")); + + // Create a subscriber + void *sub = test_context_socket (ZMQ_SUB); + TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub, "inproc://soname")); + + // Create another subscriber + void *sub2 = test_context_socket (ZMQ_SUB); + TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sub2, "inproc://soname")); + + // Subscribe for "A". + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (sub, ZMQ_SUBSCRIBE, "A", 1)); + + const uint8_t subscription[2] = {1, 'A'}; + // we must wait for the subscription to be processed here, otherwise some + // or all published messages might be lost + recv_array_expect_success (pub, subscription, 0); + + // manual subscribe message + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (pub, ZMQ_SUBSCRIBE, "A", 1)); + send_string_expect_success (pub, "A", 0); + recv_string_expect_success (sub, "A", 0); + + // Subscribe for "A". + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (sub2, ZMQ_SUBSCRIBE, "A", 1)); + recv_array_expect_success (pub, subscription, 0); + TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (pub, ZMQ_SUBSCRIBE, "A", 1)); + send_string_expect_success (pub, "A", 0); + recv_string_expect_success (sub2, "A", 0); + + char buffer[255]; + // sub won't get a message because the last subscription pipe is sub2. + TEST_ASSERT_FAILURE_ERRNO ( + EAGAIN, zmq_recv (sub, buffer, sizeof (buffer), ZMQ_DONTWAIT)); + + // Clean up. + test_context_socket_close (pub); + test_context_socket_close (sub); + test_context_socket_close (sub2); +} + +int main () +{ + setup_test_environment (); + + UNITY_BEGIN (); + RUN_TEST (test_basic); + RUN_TEST (test_unsubscribe_manual); + RUN_TEST (test_xpub_proxy_unsubscribe_on_disconnect); + RUN_TEST (test_missing_subscriptions); + RUN_TEST (test_unsubscribe_cleanup); + RUN_TEST (test_manual_last_value); + + return UNITY_END (); +}