summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--src/xpub.cpp19
-rw-r--r--src/xpub.hpp6
2 files changed, 22 insertions, 3 deletions
diff --git a/src/xpub.cpp b/src/xpub.cpp
index ed9c0e5..630450b 100644
--- a/src/xpub.cpp
+++ b/src/xpub.cpp
@@ -25,10 +25,11 @@
zmq::xpub_t::xpub_t (class ctx_t *parent_, uint32_t tid_) :
socket_base_t (parent_, tid_),
- dist (this)
+ dist (this),
+ fq (this)
{
options.type = ZMQ_XPUB;
- options.requires_in = false;
+ options.requires_in = true;
options.requires_out = true;
}
@@ -39,14 +40,16 @@ zmq::xpub_t::~xpub_t ()
void zmq::xpub_t::xattach_pipes (class reader_t *inpipe_,
class writer_t *outpipe_, const blob_t &peer_identity_)
{
- zmq_assert (!inpipe_);
+ zmq_assert (inpipe_ && outpipe_);
dist.attach (outpipe_);
+ fq.attach (inpipe_);
}
void zmq::xpub_t::process_term (int linger_)
{
// Terminate the outbound pipes.
dist.terminate ();
+ fq.terminate ();
// Continue with the termination immediately.
socket_base_t::process_term (linger_);
@@ -62,3 +65,13 @@ bool zmq::xpub_t::xhas_out ()
return dist.has_out ();
}
+int zmq::xpub_t::xrecv (zmq_msg_t *msg_, int flags_)
+{
+ return fq.recv (msg_, flags_);
+}
+
+bool zmq::xpub_t::xhas_in ()
+{
+ return fq.has_in ();
+}
+
diff --git a/src/xpub.hpp b/src/xpub.hpp
index 13baf1f..2b63ec3 100644
--- a/src/xpub.hpp
+++ b/src/xpub.hpp
@@ -24,6 +24,7 @@
#include "array.hpp"
#include "pipe.hpp"
#include "dist.hpp"
+#include "fq.hpp"
namespace zmq
{
@@ -40,6 +41,8 @@ namespace zmq
const blob_t &peer_identity_);
int xsend (zmq_msg_t *msg_, int flags_);
bool xhas_out ();
+ int xrecv (zmq_msg_t *msg_, int flags_);
+ bool xhas_in ();
private:
@@ -49,6 +52,9 @@ namespace zmq
// Distributor of messages holding the list of outbound pipes.
dist_t dist;
+ // Fair queuer for inbound subscriptions.
+ fq_t fq;
+
xpub_t (const xpub_t&);
const xpub_t &operator = (const xpub_t&);
};