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 ; +}