diff options
Diffstat (limited to 'src/xsub.cpp')
-rw-r--r-- | src/xsub.cpp | 107 |
1 files changed, 6 insertions, 101 deletions
diff --git a/src/xsub.cpp b/src/xsub.cpp index c5f610f..6928d82 100644 --- a/src/xsub.cpp +++ b/src/xsub.cpp @@ -24,19 +24,13 @@ #include "err.hpp" zmq::xsub_t::xsub_t (class ctx_t *parent_, uint32_t tid_) : - socket_base_t (parent_, tid_), - has_message (false), - more (false) + socket_base_t (parent_, tid_) { options.type = ZMQ_XSUB; - int rc = message.init (); - errno_assert (rc == 0); } zmq::xsub_t::~xsub_t () { - int rc = message.close (); - errno_assert (rc == 0); } void zmq::xsub_t::xattach_pipe (pipe_t *pipe_, const blob_t &peer_identity_) @@ -57,21 +51,8 @@ void zmq::xsub_t::xterminated (pipe_t *pipe_) int zmq::xsub_t::xsend (msg_t *msg_, int options_) { - size_t size = msg_->size (); - unsigned char *data = (unsigned char*) msg_->data (); - - // Malformed subscriptions are dropped silently. - if (size >= 1) { - - // Process a subscription. - if (*data == 1) - subscriptions.add (data + 1, size - 1); - - // Process an unsubscription. Invalid unsubscription is ignored. - if (*data == 0) - subscriptions.rm (data + 1, size - 1); - } - + // TODO: Once we'll send the subscription upstream here. For now + // just empty the message. int rc = msg_->close (); errno_assert (rc == 0); rc = msg_->init (); @@ -85,89 +66,13 @@ bool zmq::xsub_t::xhas_out () return true; } -int zmq::xsub_t::xrecv (msg_t *msg_, int flags_) +int zmq::xsub_t::xrecv (class msg_t *msg_, int flags_) { - // If there's already a message prepared by a previous call to zmq_poll, - // return it straight ahead. - if (has_message) { - int rc = msg_->move (message); - errno_assert (rc == 0); - has_message = false; - more = msg_->flags () & msg_t::more; - return 0; - } - - // TODO: This can result in infinite loop in the case of continuous - // stream of non-matching messages which breaks the non-blocking recv - // semantics. - while (true) { - - // Get a message using fair queueing algorithm. - int rc = fq.recv (msg_, flags_); - - // If there's no message available, return immediately. - // The same when error occurs. - if (rc != 0) - return -1; - - // Check whether the message matches at least one subscription. - // Non-initial parts of the message are passed - if (more || match (msg_)) { - more = msg_->flags () & msg_t::more; - return 0; - } - - // Message doesn't match. Pop any remaining parts of the message - // from the pipe. - while (msg_->flags () & msg_t::more) { - rc = fq.recv (msg_, ZMQ_DONTWAIT); - zmq_assert (rc == 0); - } - } + return fq.recv (msg_, flags_); } bool zmq::xsub_t::xhas_in () { - // There are subsequent parts of the partly-read message available. - if (more) - return true; - - // If there's already a message prepared by a previous call to zmq_poll, - // return straight ahead. - if (has_message) - return true; - - // TODO: This can result in infinite loop in the case of continuous - // stream of non-matching messages. - while (true) { - - // Get a message using fair queueing algorithm. - int rc = fq.recv (&message, ZMQ_DONTWAIT); - - // If there's no message available, return immediately. - // The same when error occurs. - if (rc != 0) { - zmq_assert (errno == EAGAIN); - return false; - } - - // Check whether the message matches at least one subscription. - if (match (&message)) { - has_message = true; - return true; - } - - // Message doesn't match. Pop any remaining parts of the message - // from the pipe. - while (message.flags () & msg_t::more) { - rc = fq.recv (&message, ZMQ_DONTWAIT); - zmq_assert (rc == 0); - } - } -} - -bool zmq::xsub_t::match (msg_t *msg_) -{ - return subscriptions.check ((unsigned char*) msg_->data (), msg_->size ()); + return fq.has_in (); } |