xpub welcome msg

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 \

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

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(&copy);		
+		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 <const int*> (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;

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 <unsigned char> blob_t;

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