Merge pull request #1267 from somdoron/manual_and_welcome

Manual and welcome

Makefile.am

@@ -328,7 +328,9 @@ 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_xpub_welcome_msg
 
 tests_test_system_SOURCES = tests/test_system.cpp
 tests_test_system_LDADD = src/libzmq.la
@@ -494,6 +496,12 @@ 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
+
+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

@@ -305,6 +305,8 @@ 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
+#define ZMQ_XPUB_WELCOME_MSG 72
 
 /*  Message options                                                           */
 #define ZMQ_MORE 1

src/xpub.cpp

@@ -28,24 +28,40 @@ 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),
+	welcome_msg ()
 {
-    options.type = ZMQ_XPUB;
+	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.
@@ -60,19 +76,28 @@ 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)) {
-            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 (size > 0 && (*data == 0 || *data == 1)) {			
+			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);
+				}
+			}
         }
         else {
             //  Process user message coming upstream from xsub socket
@@ -90,16 +115,42 @@ 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 <const int*> (optval_) < 0) {
-        errno = EINVAL;
-        return -1;
-    }
-    if (option_ == ZMQ_XPUB_VERBOSE)
-        verbose = (*static_cast <const int*> (optval_) != 0);
-    else
-    if (option_ == ZMQ_XPUB_NODROP)
-        lossy = (*static_cast <const int*> (optval_) == 0);
+{	
+	if (option_ == ZMQ_XPUB_VERBOSE || option_ == ZMQ_XPUB_NODROP || option_ == ZMQ_XPUB_MANUAL)
+	{
+		if (optvallen_ != sizeof(int) || *static_cast <const int*> (optval_) < 0) {
+			errno = EINVAL;
+			return -1;
+		}
+
+		if (option_ == ZMQ_XPUB_VERBOSE)
+			verbose = (*static_cast <const int*> (optval_) != 0);
+		else
+		if (option_ == ZMQ_XPUB_NODROP)
+			lossy = (*static_cast <const int*> (optval_) == 0);
+		else
+		if (option_ == ZMQ_XPUB_MANUAL)
+			manual = (*static_cast <const int*> (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 
+	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;
         return -1;

src/xpub.hpp

@@ -82,6 +82,15 @@ 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;
+
+		//  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_manual.cpp

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

tests/test_xpub_welcome_msg.cpp

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