From 96e29f145508f1cf4a5848c2c0d0fb83a976da0c Mon Sep 17 00:00:00 2001 From: somdoron Date: Wed, 26 Nov 2014 10:38:54 +0200 Subject: [PATCH 1/2] 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 ; +} From 768b62eb9d8434dcf30e865b56b5b8b3f946decd Mon Sep 17 00:00:00 2001 From: somdoron Date: Wed, 26 Nov 2014 13:37:36 +0200 Subject: [PATCH 2/2] xpub welcome msg --- Makefile.am | 6 ++- include/zmq.h | 1 + src/xpub.cpp | 47 +++++++++++++++------ src/xpub.hpp | 3 ++ tests/test_xpub_welcome_msg.cpp | 72 +++++++++++++++++++++++++++++++++ 5 files changed, 116 insertions(+), 13 deletions(-) create mode 100644 tests/test_xpub_welcome_msg.cpp diff --git a/Makefile.am b/Makefile.am index 00005348..ff183189 100644 --- a/Makefile.am +++ b/Makefile.am @@ -329,7 +329,8 @@ test_apps = \ tests/test_id2fd \ tests/test_capabilities \ tests/test_xpub_nodrop \ - tests/test_xpub_manual + tests/test_xpub_manual \ + tests/test_xpub_welcome_msg tests_test_system_SOURCES = tests/test_system.cpp tests_test_system_LDADD = src/libzmq.la @@ -498,6 +499,9 @@ 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 +tests_test_xpub_welcome_msg_SOURCES = tests/test_xpub_welcome_msg.cpp +tests_test_xpub_welcome_msg_LDADD = src/libzmq.la + if !ON_MINGW test_apps += \ tests/test_shutdown_stress \ diff --git a/include/zmq.h b/include/zmq.h index d542bef8..e06a35af 100644 --- a/include/zmq.h +++ b/include/zmq.h @@ -306,6 +306,7 @@ ZMQ_EXPORT const char *zmq_msg_gets (zmq_msg_t *msg, const char *property); #define ZMQ_XPUB_NODROP 69 #define ZMQ_BLOCKY 70 #define ZMQ_XPUB_MANUAL 71 +#define ZMQ_XPUB_WELCOME_MSG 72 /* Message options */ #define ZMQ_MORE 1 diff --git a/src/xpub.cpp b/src/xpub.cpp index f20cccc0..93b44347 100644 --- a/src/xpub.cpp +++ b/src/xpub.cpp @@ -29,25 +29,39 @@ zmq::xpub_t::xpub_t (class ctx_t *parent_, uint32_t tid_, int sid_) : verbose (false), more (false), lossy (true), - manual(false) + manual(false), + welcome_msg () { - last_pipe = NULL; + last_pipe = NULL; options.type = ZMQ_XPUB; + welcome_msg.init(); } zmq::xpub_t::~xpub_t () { + welcome_msg.close(); } void zmq::xpub_t::xattach_pipe (pipe_t *pipe_, bool subscribe_to_all_) { zmq_assert (pipe_); dist.attach (pipe_); - + // If subscribe_to_all_ is specified, the caller would like to subscribe // to all data on this pipe, implicitly. if (subscribe_to_all_) - subscriptions.add (NULL, 0, pipe_); + subscriptions.add (NULL, 0, pipe_); + + // if welcome message exist + if (welcome_msg.size() > 0) + { + msg_t copy; + copy.init(); + copy.copy(welcome_msg); + + pipe_->write(©); + pipe_->flush(); + } // The pipe is active when attached. Let's read the subscriptions from // it, if any. @@ -62,7 +76,7 @@ void zmq::xpub_t::xread_activated (pipe_t *pipe_) // Apply the subscription to the trie unsigned char *const data = (unsigned char *) sub.data (); const size_t size = sub.size (); - if (size > 0 && (*data == 0 || *data == 1)) { + if (size > 0 && (*data == 0 || *data == 1)) { if (manual) { last_pipe = pipe_; @@ -119,14 +133,23 @@ int zmq::xpub_t::xsetsockopt (int option_, const void *optval_, manual = (*static_cast (optval_) != 0); } else - if (option_ == ZMQ_SUBSCRIBE && manual && last_pipe != NULL) - { - subscriptions.add((unsigned char *)optval_, optvallen_, last_pipe); - } + 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); + if (option_ == ZMQ_UNSUBSCRIBE && manual && last_pipe != NULL) + subscriptions.rm((unsigned char *)optval_, optvallen_, last_pipe); + else + if (option_ == ZMQ_XPUB_WELCOME_MSG) { + welcome_msg.close(); + + if (optvallen_ > 0) { + welcome_msg.init_size(optvallen_); + + unsigned char *data = (unsigned char*)welcome_msg.data(); + memcpy(data, optval_, optvallen_); + } + else + welcome_msg.init(); } else { errno = EINVAL; diff --git a/src/xpub.hpp b/src/xpub.hpp index 50550468..8544c729 100644 --- a/src/xpub.hpp +++ b/src/xpub.hpp @@ -88,6 +88,9 @@ namespace zmq // Last pipe send subscription message, only used if xpub is on manual pipe_t *last_pipe; + // Welcome message to send to pipe when attached + msg_t welcome_msg; + // 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_welcome_msg.cpp b/tests/test_xpub_welcome_msg.cpp new file mode 100644 index 00000000..df9c9784 --- /dev/null +++ b/tests/test_xpub_welcome_msg.cpp @@ -0,0 +1,72 @@ +/* + 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 + rc = zmq_setsockopt(pub, ZMQ_XPUB_WELCOME_MSG, "W", 1); + assert (rc == 0); + + // Create a subscriber + void *sub = zmq_socket (ctx, ZMQ_SUB); + + // Subscribe to the welcome message + rc = zmq_setsockopt(sub, ZMQ_SUBSCRIBE, "W", 1); + assert(rc == 0); + + assert (sub); + rc = zmq_connect (sub, "inproc://soname"); + assert (rc == 0); + + char buffer[2]; + + // Receive the welcome subscription + rc = zmq_recv(pub, buffer, 2, 0); + assert(rc == 2); + assert(buffer[0] == 1); + assert(buffer[1] == 'W'); + + // Receive the welcome message + rc = zmq_recv(sub, buffer, 1, 0); + printf("%d\n", rc); + assert(rc == 1); + assert(buffer[0] == 'W'); + + // 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 ; +}