From 0b59866a84f733e5a53b0d2f32570581691747ef Mon Sep 17 00:00:00 2001 From: Martin Sustrik Date: Mon, 30 May 2011 10:07:34 +0200 Subject: Patches from sub-forward branch incorporated Signed-off-by: Martin Sustrik --- src/xsub.cpp | 160 ++++++++++++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 148 insertions(+), 12 deletions(-) (limited to 'src/xsub.cpp') diff --git a/src/xsub.cpp b/src/xsub.cpp index 6928d82..729f6a4 100644 --- a/src/xsub.cpp +++ b/src/xsub.cpp @@ -24,19 +24,30 @@ #include "err.hpp" zmq::xsub_t::xsub_t (class ctx_t *parent_, uint32_t tid_) : - socket_base_t (parent_, tid_) + socket_base_t (parent_, tid_), + has_message (false), + more (false) { 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_) { zmq_assert (pipe_); fq.attach (pipe_); + dist.attach (pipe_); + + // Send all the cached subscriptions to the new upstream peer. + subscriptions.apply (send_subscription, pipe_); + pipe_->flush (); } void zmq::xsub_t::xread_activated (pipe_t *pipe_) @@ -44,20 +55,49 @@ void zmq::xsub_t::xread_activated (pipe_t *pipe_) fq.activated (pipe_); } +void zmq::xsub_t::xwrite_activated (pipe_t *pipe_) +{ + dist.activated (pipe_); +} + void zmq::xsub_t::xterminated (pipe_t *pipe_) { fq.terminated (pipe_); + dist.terminated (pipe_); } -int zmq::xsub_t::xsend (msg_t *msg_, int options_) +void zmq::xsub_t::xhiccuped (pipe_t *pipe_) { - // 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 (); - errno_assert (rc == 0); - return 0; + // Send all the cached subscriptions to the hiccuped pipe. + subscriptions.apply (send_subscription, pipe_); + pipe_->flush (); +} + +int zmq::xsub_t::xsend (msg_t *msg_, int flags_) +{ + size_t size = msg_->size (); + unsigned char *data = (unsigned char*) msg_->data (); + + // Malformed subscriptions. + if (size < 1 || (*data != 0 && *data != 1)) { + errno = EINVAL; + return -1; + } + + // Process the subscription. + if (*data == 1) { + if (subscriptions.add (data + 1, size - 1)) + return dist.send (msg_, flags_); + else + return 0; + } + else if (*data == 0) { + if (subscriptions.rm (data + 1, size - 1)) + return dist.send (msg_, flags_); + else + return 0; + } + zmq_assert (false); } bool zmq::xsub_t::xhas_out () @@ -66,13 +106,109 @@ bool zmq::xsub_t::xhas_out () return true; } -int zmq::xsub_t::xrecv (class msg_t *msg_, int flags_) +int zmq::xsub_t::xrecv (msg_t *msg_, int flags_) { - return fq.recv (msg_, 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); + } + } } bool zmq::xsub_t::xhas_in () { - return fq.has_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 ()); } +void zmq::xsub_t::send_subscription (unsigned char *data_, size_t size_, + void *arg_) +{ + pipe_t *pipe = (pipe_t*) arg_; + + // Create the subsctription message. + msg_t msg; + int rc = msg.init_size (size_ + 1); + zmq_assert (rc == 0); + unsigned char *data = (unsigned char*) msg.data (); + data [0] = 1; + memcpy (data + 1, data_, size_); + + // Send it to the pipe. + bool sent = pipe->write (&msg); + zmq_assert (sent); + msg.close (); +} + + -- cgit v1.2.3