summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-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;