Merge pull request #435 from hintjens/master

Added ZMQ_XPUB_VERBOSE option

NEWS

+0MQ version 3.2.0 (RC1), released on 2012/06/05
+===============================================
+
+Bug fixes
+---------
+
+* Fixed issue 264 - Potential bug with linger, messages dropped during
+  socket close.
+
+* Fixed issue 293 - libzmq doesn't follow the ZMTP/1.0 spec (did not
+  set reserved bits to 0).
+
+* Fixed issue 303 - Assertion failure in pgm_sender.cpp:102.
+
+* Fixed issue 320 - Assertion failure in connect_session.cpp:96 when
+  connecting epgm to an invalid endpoint.
+
+* Fixed issue 325 - Assertion failure in xrep.cpp:93, when two sockets
+  connect using the same identity.
+
+* Fixed issue 327 - Assertion failure in mtrie.cpp:246, when
+  unsubscribing from channel.
+
+* Fixed issue 346 - Assertion failure in signaler.cpp:155, when using a
+  closed socket.
+
+* Fixed issue 328 - unsubscribe wrongly clears multiple subscriptions.
+
+* Fixed issue 330 - IPC listener does not remove unix domain stream file
+  when terminated.
+
+* Fixed issue 334 - Memory leak in session_base.cpp:59.
+
+* Fixed issue 369 - ROUTER cannot close/reopen while DEALER connected.
+
+Operating systems
+-----------------
+
+* Fixed issue 301 - HPUX 11iv2 - build fails, CLOCK_MONOTONIC
+  undefined.
+
+* Fixed issue 324 - OS/X - build fails, ECANTROUTE undefined.
+
+* Fixed issue 368 - Solaris / Sun C++ - build fails, no insert method
+  in multimap classes.
+
+* Fixed issue 366 - Windows - ports not freed after crash.
+
+* Fixed issue 355 - Windows - build fails, MSVC solution file is out of
+  date.
+
+* Fixed issue 331 - FreeBSD 8 and 9 - getaddrinfo fails with
+  EAI_BADFLAGS on AI_V4MAPPED flag.
+
+* Fixed issue xxx - Added support for WinCE.
+
+Performance
+-----------
+
+* Fixed issue xxx - Implemented atomic operations for ARMv7a (runs 15-20% faster).
+
+API changes
+-----------
+
+* Fixed issue 337 - Cleaned-up context API:
+
+    zmq_ctx_new() - create new context (will deprecate zmq_init)
+    zmq_ctx_destroy() - destroy context (will deprecate zmq_term)
+    zmq_ctx_set() - set context property
+    zmq_ctx_get() - get context property
+
+* Fixed issue xxx - Cleaned-up message API:
+
+    zmq_msg_send() - send a message (will deprecate zmq_sendmsg)
+    zmq_msg_recv() - receive a message (will deprecate zmq_recvmsg)
+    zmq_msg_more() - indicate whether this is final part of message
+    zmq_msg_get() - get message property
+    zmq_msg_set() - set message property
+
+* Fixed issue xxx - Added context monitoring API:
+
+    zmq_ctx_set_monitor() - configure monitor callback.
+
+* Fixed issue xxx - Added unbind/disconnect API:
+
+    zmq_unbind() - unbind socket.
+    zmq_disconnect() - disconnect socket.
+
+* Fixed issue xxx - Added ZMQ_TCP_ACCEPT_FILTER setsockopt() for listening TCP sockets.
+
+* Fixed issue 336 - Removed sys: transport.
+
+* Fixed issue 333 - Added zmq_device function back to API (was removed
+  in 3.0).
+
+* Fixed issue 340 - Add support for MAX_SOCKETS to new context API.
+
+
 OMQ version 3.1.0 (beta), released on 2011/12/18
 ================================================
 

doc/zmq_setsockopt.txt

@@ -385,6 +385,20 @@ Default value:: 0
 Applicable socket types:: ZMQ_ROUTER
 
 
+ZMQ_XPUB_VERBOSE: Set the XPUB socket behavior
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Sets the 'XPUB' socket behavior on new subscriptions. A value of '0' is the default
+and passes only new subscription messages to upstream. A value of '1' passes all
+subscription messages upstream.
+
+[horizontal]
+Option value type:: int
+Option value unit:: 0, 1
+Default value:: 0
+Applicable socket types:: ZMQ_XPUB
+
+
 ZMQ_TCP_KEEPALIVE: Override SO_KEEPALIVE socket option
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Override 'SO_KEEPALIVE' socket option(where supported by OS).

include/zmq.h

@@ -59,8 +59,8 @@ extern "C" {
 
 /*  Version macros for compile-time API version detection                     */
 #define ZMQ_VERSION_MAJOR 3
-#define ZMQ_VERSION_MINOR 3
-#define ZMQ_VERSION_PATCH 0
+#define ZMQ_VERSION_MINOR 2
+#define ZMQ_VERSION_PATCH 1
 
 #define ZMQ_MAKE_VERSION(major, minor, patch) \
     ((major) * 10000 + (minor) * 100 + (patch))
@@ -249,6 +249,7 @@ ZMQ_EXPORT int zmq_msg_set (zmq_msg_t *msg, int option, int optval);
 #define ZMQ_TCP_KEEPALIVE_INTVL 37
 #define ZMQ_TCP_ACCEPT_FILTER 38
 #define ZMQ_DELAY_ATTACH_ON_CONNECT 39
+#define ZMQ_XPUB_VERBOSE 40
 
 /*  Message options                                                           */
 #define ZMQ_MORE 1

src/xpub.cpp

@@ -28,6 +28,7 @@
 
 zmq::xpub_t::xpub_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
     socket_base_t (parent_, tid_, sid_),
+    verbose(false),
     more (false)
 {
     options.type = ZMQ_XPUB;
@@ -70,7 +71,7 @@ void zmq::xpub_t::xread_activated (pipe_t *pipe_)
 
             //  If the subscription is not a duplicate store it so that it can be
             //  passed to used on next recv call.
-            if (unique && options.type != ZMQ_PUB)
+            if (options.type == ZMQ_XPUB && (unique || verbose))
                 pending.push_back (blob_t (data, size));
         }
 
@@ -83,6 +84,21 @@ void zmq::xpub_t::xwrite_activated (pipe_t *pipe_)
     dist.activated (pipe_);
 }
 
+int zmq::xpub_t::xsetsockopt (int option_, const void *optval_,
+    size_t optvallen_)
+{
+    if (option_ != ZMQ_XPUB_VERBOSE) {
+        errno = EINVAL;
+        return -1;
+    }
+    if (optvallen_ != sizeof (int) || *static_cast <const int*> (optval_) < 0) {
+        errno = EINVAL;
+        return -1;
+    }
+    verbose = *static_cast <const int*> (optval_);
+    return 0;
+}
+
 void zmq::xpub_t::xterminated (pipe_t *pipe_)
 {
     //  Remove the pipe from the trie. If there are topics that nobody

src/xpub.hpp

@@ -54,6 +54,7 @@ namespace zmq
         bool xhas_in ();
         void xread_activated (zmq::pipe_t *pipe_);
         void xwrite_activated (zmq::pipe_t *pipe_);
+        int xsetsockopt (int option_, const void *optval_, size_t optvallen_);
         void xterminated (zmq::pipe_t *pipe_);
 
     private:
@@ -72,6 +73,10 @@ namespace zmq
         //  Distributor of messages holding the list of outbound pipes.
         dist_t dist;
 
+        // If true, send all subscription messages upstream, not just
+        // unique ones
+        bool verbose;
+
         //  True if we are in the middle of sending a multi-part message.
         bool more;