Support XPub socket send last value caching to last subscription pipe with…

 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

+# 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 <imkcy9@icloud.com>" or
+"Chengye Ke <imkcy9@gmail.com>", 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

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

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

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<const int *> (optval_) < 0) {
@@ -202,6 +208,11 @@ int zmq::xpub_t::xsetsockopt (int option_,
         } else if (option_ == ZMQ_XPUB_VERBOSER) {
             _verbose_subs = (*static_cast<const int *> (optval_) != 0);
             _verbose_unsubs = _verbose_subs;
+#ifdef ZMQ_BUILD_DRAFT_API
+        } else if (option_ == ZMQ_XPUB_MANUAL_LAST_VALUE) {
+            _manual = (*static_cast<const int *> (optval_) != 0);
+            _send_last_pipe = _manual;
+#endif
         } else if (option_ == ZMQ_XPUB_NODROP)
             _lossy = (*static_cast<const int *> (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<unsigned char *> (msg_->data ()),
+                                  msg_->size (), mark_last_pipe_as_matching,
+                                  this);
+            _last_pipe = NULL;
+        } else
+            _subscriptions.match (static_cast<unsigned char *> (msg_->data ()),
+                                  msg_->size (), mark_as_matching, this);
+#else
         _subscriptions.match (static_cast<unsigned char *> (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 ();

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;
 

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

tests/CMakeLists.txt

@@ -153,6 +153,7 @@ if(ENABLE_DRAFTS)
     test_dgram
     test_app_meta
     test_router_notify
+    test_xpub_manual_last_value
   )
 endif()
 

tests/test_xpub_manual_last_value.cpp

+/*
+    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 <http://www.gnu.org/licenses/>.
+*/
+
+#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<uint8_t> (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<uint8_t> (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 ();
+}