summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/xpub.cpp14
-rw-r--r--src/xpub.hpp4
-rw-r--r--src/xsub.cpp7
-rw-r--r--src/xsub.hpp4
4 files changed, 8 insertions, 21 deletions
diff --git a/src/xpub.cpp b/src/xpub.cpp
index 630450b..fc27144 100644
--- a/src/xpub.cpp
+++ b/src/xpub.cpp
@@ -25,11 +25,10 @@
zmq::xpub_t::xpub_t (class ctx_t *parent_, uint32_t tid_) :
socket_base_t (parent_, tid_),
- dist (this),
- fq (this)
+ dist (this)
{
options.type = ZMQ_XPUB;
- options.requires_in = true;
+ options.requires_in = false;
options.requires_out = true;
}
@@ -40,16 +39,14 @@ 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_ && outpipe_);
+ 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_);
@@ -67,11 +64,12 @@ bool zmq::xpub_t::xhas_out ()
int zmq::xpub_t::xrecv (zmq_msg_t *msg_, int flags_)
{
- return fq.recv (msg_, flags_);
+ errno = EAGAIN;
+ return -1;
}
bool zmq::xpub_t::xhas_in ()
{
- return fq.has_in ();
+ return false;
}
diff --git a/src/xpub.hpp b/src/xpub.hpp
index 2b63ec3..34da09e 100644
--- a/src/xpub.hpp
+++ b/src/xpub.hpp
@@ -24,7 +24,6 @@
#include "array.hpp"
#include "pipe.hpp"
#include "dist.hpp"
-#include "fq.hpp"
namespace zmq
{
@@ -52,9 +51,6 @@ 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&);
};
diff --git a/src/xsub.cpp b/src/xsub.cpp
index 87c8852..5eef191 100644
--- a/src/xsub.cpp
+++ b/src/xsub.cpp
@@ -27,13 +27,12 @@
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 = true;
+ options.requires_out = false;
zmq_msg_init (&message);
}
@@ -45,15 +44,13 @@ 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 e10b2c7..85f902c 100644
--- a/src/xsub.hpp
+++ b/src/xsub.hpp
@@ -24,7 +24,6 @@
#include "trie.hpp"
#include "socket_base.hpp"
-#include "dist.hpp"
#include "fq.hpp"
namespace zmq
@@ -58,9 +57,6 @@ 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;