summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMartin Sustrik <sustrik@250bpm.com>2011-01-14 12:25:54 +0100
committerMartin Sustrik <sustrik@250bpm.com>2011-01-14 12:25:54 +0100
commit58c9830d90fc39727cec88e48f8985a499aa0ee7 (patch)
tree29aa7e22b287ce13d16be0b5855f17db815714e0
parent8eae7d8507b1c96aec28bca20a157bb7537c3eb8 (diff)
XSUB socket has a subscription distributor
Signed-off-by: Martin Sustrik <sustrik@250bpm.com>
-rw-r--r--src/xsub.cpp7
-rw-r--r--src/xsub.hpp4
2 files changed, 9 insertions, 2 deletions
diff --git a/src/xsub.cpp b/src/xsub.cpp
index 5eef191..87c8852 100644
--- a/src/xsub.cpp
+++ b/src/xsub.cpp
@@ -27,12 +27,13 @@
zmq::xsub_t::xsub_t (class ctx_t *parent_, uint32_t tid_) :
socket_base_t (parent_, tid_),
fq (this),
+ dist (this),
has_message (false),
more (false)
{
options.type = ZMQ_XSUB;
options.requires_in = true;
- options.requires_out = false;
+ options.requires_out = true;
zmq_msg_init (&message);
}
@@ -44,13 +45,15 @@ zmq::xsub_t::~xsub_t ()
void zmq::xsub_t::xattach_pipes (class reader_t *inpipe_,
class writer_t *outpipe_, const blob_t &peer_identity_)
{
- zmq_assert (inpipe_ && !outpipe_);
+ zmq_assert (inpipe_ && outpipe_);
fq.attach (inpipe_);
+ dist.attach (outpipe_);
}
void zmq::xsub_t::process_term (int linger_)
{
fq.terminate ();
+ dist.terminate ();
socket_base_t::process_term (linger_);
}
diff --git a/src/xsub.hpp b/src/xsub.hpp
index 85f902c..e10b2c7 100644
--- a/src/xsub.hpp
+++ b/src/xsub.hpp
@@ -24,6 +24,7 @@
#include "trie.hpp"
#include "socket_base.hpp"
+#include "dist.hpp"
#include "fq.hpp"
namespace zmq
@@ -57,6 +58,9 @@ namespace zmq
// Fair queueing object for inbound pipes.
fq_t fq;
+ // Distributor mechanism for outbound messages (subscriptions).
+ dist_t dist;
+
// The repository of subscriptions.
trie_t subscriptions;