From 96e29f145508f1cf4a5848c2c0d0fb83a976da0c Mon Sep 17 00:00:00 2001 From: somdoron Date: Wed, 26 Nov 2014 10:38:54 +0200 Subject: [PATCH] Add manual control over subscriptions to Pub --- Makefile.am | 6 ++- include/zmq.h | 1 + src/xpub.cpp | 74 +++++++++++++++++++++++----------- src/xpub.hpp | 6 +++ tests/test_xpub_manual.cpp | 82 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 145 insertions(+), 24 deletions(-) create mode 100644 tests/test_xpub_manual.cpp diff --git a/Makefile.am b/Makefile.am index 7a1a85de..00005348 100644 --- a/Makefile.am +++ b/Makefile.am @@ -328,7 +328,8 @@ test_apps = \ tests/test_metadata \ tests/test_id2fd \ tests/test_capabilities \ - tests/test_xpub_nodrop + tests/test_xpub_nodrop \ + tests/test_xpub_manual tests_test_system_SOURCES = tests/test_system.cpp tests_test_system_LDADD = src/libzmq.la @@ -494,6 +495,9 @@ tests_test_capabilities_LDADD = src/libzmq.la tests_test_xpub_nodrop_SOURCES = tests/test_xpub_nodrop.cpp tests_test_xpub_nodrop_LDADD = src/libzmq.la +tests_test_xpub_manual_SOURCES = tests/test_xpub_manual.cpp +tests_test_xpub_manual_LDADD = src/libzmq.la + if !ON_MINGW test_apps += \ tests/test_shutdown_stress \ diff --git a/include/zmq.h b/include/zmq.h index 199e4cba..d542bef8 100644 --- a/include/zmq.h +++ b/include/zmq.h @@ -305,6 +305,7 @@ ZMQ_EXPORT const char *zmq_msg_gets (zmq_msg_t *msg, const char *property); #define ZMQ_SOCKS_PROXY 68 #define ZMQ_XPUB_NODROP 69 #define ZMQ_BLOCKY 70 +#define ZMQ_XPUB_MANUAL 71 /* Message options */ #define ZMQ_MORE 1 diff --git a/src/xpub.cpp b/src/xpub.cpp index 7c9bcaea..f20cccc0 100644 --- a/src/xpub.cpp +++ b/src/xpub.cpp @@ -28,9 +28,11 @@ zmq::xpub_t::xpub_t (class ctx_t *parent_, uint32_t tid_, int sid_) : socket_base_t (parent_, tid_, sid_), verbose (false), more (false), - lossy (true) + lossy (true), + manual(false) { - options.type = ZMQ_XPUB; + last_pipe = NULL; + options.type = ZMQ_XPUB; } zmq::xpub_t::~xpub_t () @@ -61,18 +63,27 @@ void zmq::xpub_t::xread_activated (pipe_t *pipe_) unsigned char *const data = (unsigned char *) sub.data (); const size_t size = sub.size (); if (size > 0 && (*data == 0 || *data == 1)) { - bool unique; - if (*data == 0) - unique = subscriptions.rm (data + 1, size - 1, pipe_); - else - unique = subscriptions.add (data + 1, size - 1, pipe_); + if (manual) + { + last_pipe = pipe_; + pending_data.push_back(blob_t(data, size)); + pending_flags.push_back(0); + } + else + { + bool unique; + if (*data == 0) + unique = subscriptions.rm(data + 1, size - 1, pipe_); + else + unique = subscriptions.add(data + 1, size - 1, pipe_); - // If the subscription is not a duplicate store it so that it can be - // passed to used on next recv call. (Unsubscribe is not verbose.) - if (options.type == ZMQ_XPUB && (unique || (*data && verbose))) { - pending_data.push_back (blob_t (data, size)); - pending_flags.push_back (0); - } + // If the subscription is not a duplicate store it so that it can be + // passed to used on next recv call. (Unsubscribe is not verbose.) + if (options.type == ZMQ_XPUB && (unique || (*data && verbose))) { + pending_data.push_back(blob_t(data, size)); + pending_flags.push_back(0); + } + } } else { // Process user message coming upstream from xsub socket @@ -90,16 +101,33 @@ void zmq::xpub_t::xwrite_activated (pipe_t *pipe_) int zmq::xpub_t::xsetsockopt (int option_, const void *optval_, size_t optvallen_) -{ - if (optvallen_ != sizeof (int) || *static_cast (optval_) < 0) { - errno = EINVAL; - return -1; - } - if (option_ == ZMQ_XPUB_VERBOSE) - verbose = (*static_cast (optval_) != 0); - else - if (option_ == ZMQ_XPUB_NODROP) - lossy = (*static_cast (optval_) == 0); +{ + if (option_ == ZMQ_XPUB_VERBOSE || option_ == ZMQ_XPUB_NODROP || option_ == ZMQ_XPUB_MANUAL) + { + if (optvallen_ != sizeof(int) || *static_cast (optval_) < 0) { + errno = EINVAL; + return -1; + } + + if (option_ == ZMQ_XPUB_VERBOSE) + verbose = (*static_cast (optval_) != 0); + else + if (option_ == ZMQ_XPUB_NODROP) + lossy = (*static_cast (optval_) == 0); + else + if (option_ == ZMQ_XPUB_MANUAL) + manual = (*static_cast (optval_) != 0); + } + else + if (option_ == ZMQ_SUBSCRIBE && manual && last_pipe != NULL) + { + subscriptions.add((unsigned char *)optval_, optvallen_, last_pipe); + } + else + if (option_ == ZMQ_UNSUBSCRIBE && manual && last_pipe != NULL) + { + subscriptions.rm((unsigned char *)optval_, optvallen_, last_pipe); + } else { errno = EINVAL; return -1; diff --git a/src/xpub.hpp b/src/xpub.hpp index 89c7fa4b..50550468 100644 --- a/src/xpub.hpp +++ b/src/xpub.hpp @@ -82,6 +82,12 @@ namespace zmq // Drop messages if HWM reached, otherwise return with EAGAIN bool lossy; + // Subscriptions will not bed added automatically, only after calling set option with ZMQ_SUBSCRIBE or ZMQ_UNSUBSCRIBE + bool manual; + + // Last pipe send subscription message, only used if xpub is on manual + pipe_t *last_pipe; + // List of pending (un)subscriptions, ie. those that were already // applied to the trie, but not yet received by the user. typedef std::basic_string blob_t; diff --git a/tests/test_xpub_manual.cpp b/tests/test_xpub_manual.cpp new file mode 100644 index 00000000..d8f10349 --- /dev/null +++ b/tests/test_xpub_manual.cpp @@ -0,0 +1,82 @@ +/* + Copyright (c) 2007-2014 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 . +*/ + +#include "testutil.hpp" + +int main (void) +{ + setup_test_environment(); + void *ctx = zmq_ctx_new (); + assert (ctx); + + // Create a publisher + void *pub = zmq_socket (ctx, ZMQ_XPUB); + assert (pub); + int rc = zmq_bind (pub, "inproc://soname"); + assert (rc == 0); + + // set pub socket options + int manual = 1; + rc = zmq_setsockopt(pub, ZMQ_XPUB_MANUAL, &manual, 4); + assert (rc == 0); + + // Create a subscriber + void *sub = zmq_socket (ctx, ZMQ_XSUB); + assert (sub); + rc = zmq_connect (sub, "inproc://soname"); + assert (rc == 0); + + // Subscribe for A + char subscription[2] = { 1, 'A'}; + rc = zmq_send_const(sub, subscription, 2, 0); + assert (rc == 2); + + char buffer[2]; + + // Receive subscriptions from subscriber + rc = zmq_recv(pub, buffer, 2, 0); + assert(rc == 2); + assert(buffer[0] == 1); + assert(buffer[1] == 'A'); + + // Subscribe socket for B instead + rc = zmq_setsockopt(pub, ZMQ_SUBSCRIBE, "B", 1); + assert(rc == 0); + + // Sending A message and B Message + rc = zmq_send_const(pub, "A", 1, 0); + assert(rc == 1); + + rc = zmq_send_const(pub, "B", 1, 0); + assert(rc == 1); + + rc = zmq_recv(sub, buffer, 1, ZMQ_DONTWAIT); + assert(rc == 1); + assert(buffer[0] == 'B'); + + // Clean up. + rc = zmq_close (pub); + assert (rc == 0); + rc = zmq_close (sub); + assert (rc == 0); + rc = zmq_ctx_term (ctx); + assert (rc == 0); + + return 0 ; +}