generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-10-14 23:07:59 +02:00
Code Issues Projects Releases Wiki Activity

Fix 1478: receive unsubscriptions in XPUB when verbose

Browse Source 
Fixes not receiving unsubscription messages in XPUB socket with
ZMQ_XPUB_VERBOSE and using a XSUB-XPUB proxy in front.

This adds two modifications:

- It adds a new flag, ZMQ_XPUB_VERBOSE_UNSUBSCRIBE, to enable verbose
  unsubscription messages, necessary when using a XSUB/XPUB proxy.

- It adds a boolean switch to zmq::mtrie_t::rm () to control if the
  callback is invoked every time or only in the last removal. Necessary
  when a pipe is terminated and the verbose mode for unsubscriptions is
  enabled.
This commit is contained in:
Ricardo Catalinas Jiménez
2015-07-21 23:35:50 +01:00
parent 305c07583d
commit ec5592db1f
6 changed files with 61 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/xpub.cpp
View File

@@ -36,7 +36,8 @@
zmq::xpub_t::xpub_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
socket_base_t (parent_, tid_, sid_),
verbose (false),
verbose_subs (false),
verbose_unsubs (false),
more (false),
lossy (true),
manual(false),
@@ -101,9 +102,11 @@ void zmq::xpub_t::xread_activated (pipe_t *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))) {
// If the (un)subscription is not a duplicate store it so that it can be
// passed to the user on next recv call unless verbose mode is enabled
// which makes to pass always these messages.
if (options.type == ZMQ_XPUB && (unique || (*data == 1 && verbose_subs) ||
(*data == 0 && verbose_unsubs && verbose_subs))) {
pending_data.push_back(blob_t(data, size));
pending_flags.push_back(0);
}
@@ -126,7 +129,8 @@ void zmq::xpub_t::xwrite_activated (pipe_t *pipe_)
int zmq::xpub_t::xsetsockopt (int option_, const void *optval_,
size_t optvallen_)
{
if (option_ == ZMQ_XPUB_VERBOSE || option_ == ZMQ_XPUB_NODROP || option_ == ZMQ_XPUB_MANUAL)
if (option_ == ZMQ_XPUB_VERBOSE || option_ == ZMQ_XPUB_VERBOSE_UNSUBSCRIBE ||
option_ == ZMQ_XPUB_NODROP || option_ == ZMQ_XPUB_MANUAL)
{
if (optvallen_ != sizeof(int) || *static_cast <const int*> (optval_) < 0) {
errno = EINVAL;
@@ -134,7 +138,10 @@ int zmq::xpub_t::xsetsockopt (int option_, const void *optval_,
}
if (option_ == ZMQ_XPUB_VERBOSE)
verbose = (*static_cast <const int*> (optval_) != 0);
verbose_subs = (*static_cast <const int*> (optval_) != 0);
else
if (option_ == ZMQ_XPUB_VERBOSE_UNSUBSCRIBE)
verbose_unsubs = (*static_cast <const int*> (optval_) != 0);
else
if (option_ == ZMQ_XPUB_NODROP)
lossy = (*static_cast <const int*> (optval_) == 0);
@@ -173,7 +180,7 @@ void zmq::xpub_t::xpipe_terminated (pipe_t *pipe_)
// 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);
subscriptions.rm (pipe_, send_unsubscription, this, !verbose_unsubs);
dist.pipe_terminated (pipe_);
}