Add manual control over subscriptions to Pub

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 \

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

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 (manual)
-            if (*data == 0)
+			{
-                unique = subscriptions.rm (data + 1, size - 1, pipe_);
+				last_pipe = pipe_;
-            else
+				pending_data.push_back(blob_t(data, size));
-                unique = subscriptions.add (data + 1, size - 1, pipe_);
+				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
+				//  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.)
+				//  passed to used on next recv call. (Unsubscribe is not verbose.)
-            if (options.type == ZMQ_XPUB && (unique || (*data && verbose))) {
+				if (options.type == ZMQ_XPUB && (unique || (*data && verbose))) {
-                pending_data.push_back (blob_t (data, size));
+					pending_data.push_back(blob_t(data, size));
-                pending_flags.push_back (0);
+					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 <const int*> (optval_) < 0) {
+	if (option_ == ZMQ_XPUB_VERBOSE || option_ == ZMQ_XPUB_NODROP || option_ == ZMQ_XPUB_MANUAL)
-        errno = EINVAL;
+	{
-        return -1;
+		if (optvallen_ != sizeof(int) || *static_cast <const int*> (optval_) < 0) {
-    }
+			errno = EINVAL;
-    if (option_ == ZMQ_XPUB_VERBOSE)
+			return -1;
-        verbose = (*static_cast <const int*> (optval_) != 0);
+		}
-    else
+
-    if (option_ == ZMQ_XPUB_NODROP)
+		if (option_ == ZMQ_XPUB_VERBOSE)
-        lossy = (*static_cast <const int*> (optval_) == 0);
+			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 {
         errno = EINVAL;
         return -1;

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

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