Store manual subscriptions in XPUB and send them out on pipe termination.

src/xpub.cpp

@@ -92,6 +92,12 @@ void zmq::xpub_t::xread_activated (pipe_t *pipe_)
         if (size > 0 && (*data == 0 || *data == 1)) {
             if (manual)
             {
+                // Store manual subscription to use on termination
+                if (*data == 0)
+                    manual_subscriptions.rm(data + 1, size - 1, pipe_);
+                else
+                    manual_subscriptions.add(data + 1, size - 1, pipe_);
+
                 pending_pipes.push_back(pipe_);
                 pending_data.push_back(blob_t(data, size));
                 pending_metadata.push_back(sub.metadata());
@@ -190,11 +196,20 @@ int zmq::xpub_t::xsetsockopt (int option_, const void *optval_,
 }
 
 void zmq::xpub_t::xpipe_terminated (pipe_t *pipe_)
+{
+    if (manual)
+    {
+        //  Remove the pipe from the trie and send corresponding manual
+        //  unsubscriptions upstream.
+        manual_subscriptions.rm (pipe_, send_unsubscription, this, false);
+    }
+    else
     {
         //  Remove the pipe from the trie. If there are topics that nobody
         //  is interested in anymore, send corresponding unsubscriptions
         //  upstream.
-    subscriptions.rm (pipe_, send_unsubscription, this, !(verbose_unsubs || manual));
+        subscriptions.rm (pipe_, send_unsubscription, this, !verbose_unsubs);
+    }
 
     dist.pipe_terminated (pipe_);
 }

src/xpub.hpp

@@ -79,6 +79,9 @@ namespace zmq
         //  List of all subscriptions mapped to corresponding pipes.
         mtrie_t subscriptions;
 
+        //  List of manual subscriptions mapped to corresponding pipes.
+        mtrie_t manual_subscriptions;
+
         //  Distributor of messages holding the list of outbound pipes.
         dist_t dist;