xsub.cpp 7.96 KB
Newer Older
1
/*
2
    Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file
3

4
    This file is part of libzmq, the ZeroMQ core engine in C++.
5

6 7 8
    libzmq is free software; you can redistribute it and/or modify it under
    the terms of the GNU Lesser General Public License (LGPL) as published
    by the Free Software Foundation; either version 3 of the License, or
9 10
    (at your option) any later version.

11 12 13 14 15 16 17 18 19 20 21 22 23 24
    As a special exception, the Contributors give you permission to link
    this library with independent modules to produce an executable,
    regardless of the license terms of these independent modules, and to
    copy and distribute the resulting executable under terms of your choice,
    provided that you also meet, for each linked independent module, the
    terms and conditions of the license of that module. An independent
    module is a module which is not derived from or based on this library.
    If you modify this library, you must extend this exception to your
    version of the library.

    libzmq 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.
25 26 27 28 29

    You should have received a copy of the GNU Lesser General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

30
#include "precompiled.hpp"
31 32
#include <string.h>

33
#include "macros.hpp"
34 35 36
#include "xsub.hpp"
#include "err.hpp"

37 38
zmq::xsub_t::xsub_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
    socket_base_t (parent_, tid_, sid_),
39 40
    _has_message (false),
    _more (false)
41
{
42
    options.type = ZMQ_XSUB;
43 44 45

    //  When socket is being closed down we don't want to wait till pending
    //  subscription commands are sent to the wire.
46
    options.linger.store (0);
47

48
    int rc = _message.init ();
49
    errno_assert (rc == 0);
50 51 52 53
}

zmq::xsub_t::~xsub_t ()
{
54
    int rc = _message.close ();
55
    errno_assert (rc == 0);
56 57
}

58 59 60
void zmq::xsub_t::xattach_pipe (pipe_t *pipe_,
                                bool subscribe_to_all_,
                                bool locally_initiated_)
61
{
62
    LIBZMQ_UNUSED (subscribe_to_all_);
63
    LIBZMQ_UNUSED (locally_initiated_);
64

65
    zmq_assert (pipe_);
66 67
    _fq.attach (pipe_);
    _dist.attach (pipe_);
68 69

    //  Send all the cached subscriptions to the new upstream peer.
70
    _subscriptions.apply (send_subscription, pipe_);
71
    pipe_->flush ();
72 73
}

74
void zmq::xsub_t::xread_activated (pipe_t *pipe_)
75
{
76
    _fq.activated (pipe_);
77 78
}

79 80
void zmq::xsub_t::xwrite_activated (pipe_t *pipe_)
{
81
    _dist.activated (pipe_);
82 83
}

84
void zmq::xsub_t::xpipe_terminated (pipe_t *pipe_)
85
{
86 87
    _fq.pipe_terminated (pipe_);
    _dist.pipe_terminated (pipe_);
88 89
}

90
void zmq::xsub_t::xhiccuped (pipe_t *pipe_)
91
{
92
    //  Send all the cached subscriptions to the hiccuped pipe.
93
    _subscriptions.apply (send_subscription, pipe_);
94 95 96
    pipe_->flush ();
}

97
int zmq::xsub_t::xsend (msg_t *msg_)
98 99
{
    size_t size = msg_->size ();
100
    unsigned char *data = static_cast<unsigned char *> (msg_->data ());
101

102
    if (msg_->is_subscribe () || (size > 0 && *data == 1)) {
103
        //  Process subscribe message
104 105 106 107
        //  This used to filter out duplicate subscriptions,
        //  however this is alread done on the XPUB side and
        //  doing it here as well breaks ZMQ_XPUB_VERBOSE
        //  when there are forwarding devices involved.
108 109 110 111 112 113 114 115
        if (msg_->is_subscribe ()) {
            data = static_cast<unsigned char *> (msg_->command_body ());
            size = msg_->command_body_size ();
        } else {
            data = data + 1;
            size = size - 1;
        }
        _subscriptions.add (data, size);
116
        return _dist.send_to_all (msg_);
117
    }
118
    if (msg_->is_cancel () || (size > 0 && *data == 0)) {
119
        //  Process unsubscribe message
120 121 122 123 124 125 126 127
        if (msg_->is_cancel ()) {
            data = static_cast<unsigned char *> (msg_->command_body ());
            size = msg_->command_body_size ();
        } else {
            data = data + 1;
            size = size - 1;
        }
        if (_subscriptions.rm (data, size))
128
            return _dist.send_to_all (msg_);
129
    } else
130
        //  User message sent upstream to XPUB socket
131
        return _dist.send_to_all (msg_);
132

Martin Hurton's avatar
Martin Hurton committed
133 134 135 136 137 138
    int rc = msg_->close ();
    errno_assert (rc == 0);
    rc = msg_->init ();
    errno_assert (rc == 0);

    return 0;
139 140 141 142 143 144
}

bool zmq::xsub_t::xhas_out ()
{
    //  Subscription can be added/removed anytime.
    return true;
145 146
}

147
int zmq::xsub_t::xrecv (msg_t *msg_)
148
{
149 150
    //  If there's already a message prepared by a previous call to zmq_poll,
    //  return it straight ahead.
151 152
    if (_has_message) {
        int rc = msg_->move (_message);
153
        errno_assert (rc == 0);
154 155
        _has_message = false;
        _more = (msg_->flags () & msg_t::more) != 0;
156 157 158 159 160 161 162 163
        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.
164
        int rc = _fq.recv (msg_);
165 166 167 168 169 170 171

        //  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.
172
        //  Non-initial parts of the message are passed
173 174
        if (_more || !options.filter || match (msg_)) {
            _more = (msg_->flags () & msg_t::more) != 0;
175 176 177 178 179
            return 0;
        }

        //  Message doesn't match. Pop any remaining parts of the message
        //  from the pipe.
180
        while (msg_->flags () & msg_t::more) {
181
            rc = _fq.recv (msg_);
182
            errno_assert (rc == 0);
183 184
        }
    }
185 186 187 188
}

bool zmq::xsub_t::xhas_in ()
{
189
    //  There are subsequent parts of the partly-read message available.
190
    if (_more)
191 192 193 194
        return true;

    //  If there's already a message prepared by a previous call to zmq_poll,
    //  return straight ahead.
195
    if (_has_message)
196 197 198 199 200 201
        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.
202
        int rc = _fq.recv (&_message);
203 204 205 206

        //  If there's no message available, return immediately.
        //  The same when error occurs.
        if (rc != 0) {
207
            errno_assert (errno == EAGAIN);
208 209 210 211
            return false;
        }

        //  Check whether the message matches at least one subscription.
212 213
        if (!options.filter || match (&_message)) {
            _has_message = true;
214 215 216 217 218
            return true;
        }

        //  Message doesn't match. Pop any remaining parts of the message
        //  from the pipe.
219 220
        while (_message.flags () & msg_t::more) {
            rc = _fq.recv (&_message);
221
            errno_assert (rc == 0);
222 223 224 225 226 227
        }
    }
}

bool zmq::xsub_t::match (msg_t *msg_)
{
228
    bool matching = _subscriptions.check (
229
      static_cast<unsigned char *> (msg_->data ()), msg_->size ());
230 231

    return matching ^ options.invert_matching;
232
}
233

234 235 236
void zmq::xsub_t::send_subscription (unsigned char *data_,
                                     size_t size_,
                                     void *arg_)
237
{
238
    pipe_t *pipe = static_cast<pipe_t *> (arg_);
239

240
    //  Create the subscription message.
241 242
    msg_t msg;
    int rc = msg.init_size (size_ + 1);
243
    errno_assert (rc == 0);
244
    unsigned char *data = static_cast<unsigned char *> (msg.data ());
245
    data[0] = 1;
246 247 248 249

    //  We explicitly allow a NULL subscription with size zero
    if (size_) {
        assert (data_);
250
        memcpy (data + 1, data_, size_);
251
    }
252 253 254

    //  Send it to the pipe.
    bool sent = pipe->write (&msg);
255 256 257 258 259 260
    //  If we reached the SNDHWM, and thus cannot send the subscription, drop
    //  the subscription message instead. This matches the behaviour of
    //  zmq_setsockopt(ZMQ_SUBSCRIBE, ...), which also drops subscriptions
    //  when the SNDHWM is reached.
    if (!sent)
        msg.close ();
261
}