diff options
Diffstat (limited to 'src/xsub.cpp')
| -rw-r--r-- | src/xsub.cpp | 62 |
1 files changed, 31 insertions, 31 deletions
diff --git a/src/xsub.cpp b/src/xsub.cpp index e227c75..edde39e 100644 --- a/src/xsub.cpp +++ b/src/xsub.cpp @@ -3,14 +3,14 @@ Copyright (c) 2011 VMware, Inc. Copyright (c) 2010-2011 Other contributors as noted in the AUTHORS file - This file is part of 0MQ. + This file is part of Crossroads project. - 0MQ is free software; you can redistribute it and/or modify it under + Crossroads is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published by the Free Software Foundation; either version 3 of the License, or (at your option) any later version. - 0MQ is distributed in the hope that it will be useful, + Crossroads is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details. @@ -24,12 +24,12 @@ #include "xsub.hpp" #include "err.hpp" -zmq::xsub_t::xsub_t (class ctx_t *parent_, uint32_t tid_, int sid_) : +xs::xsub_t::xsub_t (class ctx_t *parent_, uint32_t tid_, int sid_) : socket_base_t (parent_, tid_, sid_), has_message (false), more (false) { - options.type = ZMQ_XSUB; + options.type = XS_XSUB; // When socket is being closed down we don't want to wait till pending // subscription commands are sent to the wire. @@ -39,15 +39,15 @@ zmq::xsub_t::xsub_t (class ctx_t *parent_, uint32_t tid_, int sid_) : errno_assert (rc == 0); } -zmq::xsub_t::~xsub_t () +xs::xsub_t::~xsub_t () { int rc = message.close (); errno_assert (rc == 0); } -void zmq::xsub_t::xattach_pipe (pipe_t *pipe_) +void xs::xsub_t::xattach_pipe (pipe_t *pipe_) { - zmq_assert (pipe_); + xs_assert (pipe_); fq.attach (pipe_); dist.attach (pipe_); @@ -56,30 +56,30 @@ void zmq::xsub_t::xattach_pipe (pipe_t *pipe_) pipe_->flush (); } -void zmq::xsub_t::xread_activated (pipe_t *pipe_) +void xs::xsub_t::xread_activated (pipe_t *pipe_) { fq.activated (pipe_); } -void zmq::xsub_t::xwrite_activated (pipe_t *pipe_) +void xs::xsub_t::xwrite_activated (pipe_t *pipe_) { dist.activated (pipe_); } -void zmq::xsub_t::xterminated (pipe_t *pipe_) +void xs::xsub_t::xterminated (pipe_t *pipe_) { fq.terminated (pipe_); dist.terminated (pipe_); } -void zmq::xsub_t::xhiccuped (pipe_t *pipe_) +void xs::xsub_t::xhiccuped (pipe_t *pipe_) { // 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_) +int xs::xsub_t::xsend (msg_t *msg_, int flags_) { size_t size = msg_->size (); unsigned char *data = (unsigned char*) msg_->data (); @@ -104,19 +104,19 @@ int zmq::xsub_t::xsend (msg_t *msg_, int flags_) return 0; } - zmq_assert (false); + xs_assert (false); return -1; } -bool zmq::xsub_t::xhas_out () +bool xs::xsub_t::xhas_out () { // Subscription can be added/removed anytime. return true; } -int zmq::xsub_t::xrecv (msg_t *msg_, int flags_) +int xs::xsub_t::xrecv (msg_t *msg_, int flags_) { - // If there's already a message prepared by a previous call to zmq_poll, + // If there's already a message prepared by a previous call to xs_poll, // return it straight ahead. if (has_message) { int rc = msg_->move (message); @@ -149,19 +149,19 @@ int zmq::xsub_t::xrecv (msg_t *msg_, int flags_) // 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); + rc = fq.recv (msg_, XS_DONTWAIT); + xs_assert (rc == 0); } } } -bool zmq::xsub_t::xhas_in () +bool xs::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, + // If there's already a message prepared by a previous call to xs_poll, // return straight ahead. if (has_message) return true; @@ -171,12 +171,12 @@ bool zmq::xsub_t::xhas_in () while (true) { // Get a message using fair queueing algorithm. - int rc = fq.recv (&message, ZMQ_DONTWAIT); + int rc = fq.recv (&message, XS_DONTWAIT); // If there's no message available, return immediately. // The same when error occurs. if (rc != 0) { - zmq_assert (errno == EAGAIN); + xs_assert (errno == EAGAIN); return false; } @@ -189,18 +189,18 @@ bool zmq::xsub_t::xhas_in () // 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); + rc = fq.recv (&message, XS_DONTWAIT); + xs_assert (rc == 0); } } } -bool zmq::xsub_t::match (msg_t *msg_) +bool xs::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 xs::xsub_t::send_subscription (unsigned char *data_, size_t size_, void *arg_) { pipe_t *pipe = (pipe_t*) arg_; @@ -208,17 +208,17 @@ void zmq::xsub_t::send_subscription (unsigned char *data_, size_t size_, // Create the subsctription message. msg_t msg; int rc = msg.init_size (size_ + 1); - zmq_assert (rc == 0); + xs_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); + xs_assert (sent); } -zmq::xsub_session_t::xsub_session_t (io_thread_t *io_thread_, bool connect_, +xs::xsub_session_t::xsub_session_t (io_thread_t *io_thread_, bool connect_, socket_base_t *socket_, const options_t &options_, const char *protocol_, const char *address_) : session_base_t (io_thread_, connect_, socket_, options_, protocol_, @@ -226,7 +226,7 @@ zmq::xsub_session_t::xsub_session_t (io_thread_t *io_thread_, bool connect_, { } -zmq::xsub_session_t::~xsub_session_t () +xs::xsub_session_t::~xsub_session_t () { } |