xpub.cpp 12.1 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 34
#include "xpub.hpp"
#include "pipe.hpp"
35 36
#include "err.hpp"
#include "msg.hpp"
37
#include "macros.hpp"
38
#include "generic_mtrie_impl.hpp"
39

40 41
zmq::xpub_t::xpub_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
    socket_base_t (parent_, tid_, sid_),
42 43 44 45 46 47 48
    _verbose_subs (false),
    _verbose_unsubs (false),
    _more (false),
    _lossy (true),
    _manual (false),
    _pending_pipes (),
    _welcome_msg ()
49
{
50
    _last_pipe = NULL;
51
    options.type = ZMQ_XPUB;
52
    _welcome_msg.init ();
53 54 55 56
}

zmq::xpub_t::~xpub_t ()
{
57
    _welcome_msg.close ();
58 59
}

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

66
    zmq_assert (pipe_);
67
    _dist.attach (pipe_);
68

69
    //  If subscribe_to_all_ is specified, the caller would like to subscribe
70
    //  to all data on this pipe, implicitly.
71
    if (subscribe_to_all_)
72
        _subscriptions.add (NULL, 0, pipe_);
somdoron's avatar
somdoron committed
73

74
    // if welcome message exists, send a copy of it
75
    if (_welcome_msg.size () > 0) {
76
        msg_t copy;
77
        copy.init ();
78
        int rc = copy.copy (_welcome_msg);
79 80 81 82
        errno_assert (rc == 0);
        bool ok = pipe_->write (&copy);
        zmq_assert (ok);
        pipe_->flush ();
83
    }
84

85 86 87
    //  The pipe is active when attached. Let's read the subscriptions from
    //  it, if any.
    xread_activated (pipe_);
88 89 90 91
}

void zmq::xpub_t::xread_activated (pipe_t *pipe_)
{
92 93
    //  There are some subscriptions waiting. Let's process them.
    msg_t sub;
94
    while (pipe_->read (&sub)) {
95
        metadata_t *metadata = sub.metadata ();
96 97 98 99
        unsigned char *msg_data = static_cast<unsigned char *> (sub.data ()),
                      *data = NULL;
        size_t size = 0;
        bool subscribe = false;
100

101 102 103 104 105 106 107 108 109 110 111
        //  Apply the subscription to the trie
        if (sub.is_subscribe () || sub.is_cancel ()) {
            data = static_cast<unsigned char *> (sub.command_body ());
            size = sub.command_body_size ();
            subscribe = sub.is_subscribe ();
        } else if (sub.size () > 0) {
            unsigned char first = *msg_data;
            if (first == 0 || first == 1) {
                data = msg_data + 1;
                size = sub.size () - 1;
                subscribe = first == 1;
112
            }
113
        } else {
114
            //  Process user message coming upstream from xsub socket
115
            _pending_data.push_back (blob_t (msg_data, sub.size ()));
116
            if (metadata)
117
                metadata->add_ref ();
118 119
            _pending_metadata.push_back (metadata);
            _pending_flags.push_back (sub.flags ());
120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177
            sub.close ();
            continue;
        }

        if (_manual) {
            // Store manual subscription to use on termination
            if (!subscribe)
                _manual_subscriptions.rm (data, size, pipe_);
            else
                _manual_subscriptions.add (data, size, pipe_);

            _pending_pipes.push_back (pipe_);

            //  ZMTP 3.1 hack: we need to support sub/cancel commands, but
            //  we can't give them back to userspace as it would be an API
            //  breakage since the payload of the message is completely
            //  different. Manually craft an old-style message instead.
            data = data - 1;
            size = size + 1;
            if (subscribe)
                *data = 1;
            else
                *data = 0;

            _pending_data.push_back (blob_t (data, size));
            if (metadata)
                metadata->add_ref ();
            _pending_metadata.push_back (metadata);
            _pending_flags.push_back (0);
        } else {
            bool notify;
            if (!subscribe) {
                mtrie_t::rm_result rm_result =
                  _subscriptions.rm (data, size, pipe_);
                //  TODO reconsider what to do if rm_result == mtrie_t::not_found
                notify = rm_result != mtrie_t::values_remain || _verbose_unsubs;
            } else {
                bool first_added = _subscriptions.add (data, size, pipe_);
                notify = first_added || _verbose_subs;
            }

            //  If the request was a new subscription, or the subscription
            //  was removed, or verbose mode is enabled, store it so that
            //  it can be passed to the user on next recv call.
            if (options.type == ZMQ_XPUB && notify) {
                data = data - 1;
                size = size + 1;
                if (subscribe)
                    *data = 1;
                else
                    *data = 0;

                _pending_data.push_back (blob_t (data, size));
                if (metadata)
                    metadata->add_ref ();
                _pending_metadata.push_back (metadata);
                _pending_flags.push_back (0);
            }
178
        }
179
        sub.close ();
180
    }
181 182
}

183
void zmq::xpub_t::xwrite_activated (pipe_t *pipe_)
184
{
185
    _dist.activated (pipe_);
186 187
}

188 189 190
int zmq::xpub_t::xsetsockopt (int option_,
                              const void *optval_,
                              size_t optvallen_)
191
{
192 193 194 195
    if (option_ == ZMQ_XPUB_VERBOSE || option_ == ZMQ_XPUB_VERBOSER
        || option_ == ZMQ_XPUB_NODROP || option_ == ZMQ_XPUB_MANUAL) {
        if (optvallen_ != sizeof (int)
            || *static_cast<const int *> (optval_) < 0) {
196 197 198
            errno = EINVAL;
            return -1;
        }
199
        if (option_ == ZMQ_XPUB_VERBOSE) {
200 201
            _verbose_subs = (*static_cast<const int *> (optval_) != 0);
            _verbose_unsubs = false;
202
        } else if (option_ == ZMQ_XPUB_VERBOSER) {
203 204
            _verbose_subs = (*static_cast<const int *> (optval_) != 0);
            _verbose_unsubs = _verbose_subs;
205
        } else if (option_ == ZMQ_XPUB_NODROP)
206
            _lossy = (*static_cast<const int *> (optval_) == 0);
207
        else if (option_ == ZMQ_XPUB_MANUAL)
208 209 210 211 212 213 214 215 216
            _manual = (*static_cast<const int *> (optval_) != 0);
    } else if (option_ == ZMQ_SUBSCRIBE && _manual) {
        if (_last_pipe != NULL)
            _subscriptions.add ((unsigned char *) optval_, optvallen_,
                                _last_pipe);
    } else if (option_ == ZMQ_UNSUBSCRIBE && _manual) {
        if (_last_pipe != NULL)
            _subscriptions.rm ((unsigned char *) optval_, optvallen_,
                               _last_pipe);
217
    } else if (option_ == ZMQ_XPUB_WELCOME_MSG) {
218
        _welcome_msg.close ();
219 220

        if (optvallen_ > 0) {
221
            int rc = _welcome_msg.init_size (optvallen_);
222
            errno_assert (rc == 0);
223

224
            unsigned char *data =
225
              static_cast<unsigned char *> (_welcome_msg.data ());
226
            memcpy (data, optval_, optvallen_);
227
        } else
228
            _welcome_msg.init ();
229
    } else {
230 231 232
        errno = EINVAL;
        return -1;
    }
Pieter Hintjens's avatar
Pieter Hintjens committed
233 234 235
    return 0;
}

236
static void stub (zmq::mtrie_t::prefix_t data_, size_t size_, void *arg_)
237
{
238 239 240
    LIBZMQ_UNUSED (data_);
    LIBZMQ_UNUSED (size_);
    LIBZMQ_UNUSED (arg_);
241 242
}

243
void zmq::xpub_t::xpipe_terminated (pipe_t *pipe_)
244
{
245
    if (_manual) {
246 247
        //  Remove the pipe from the trie and send corresponding manual
        //  unsubscriptions upstream.
248
        _manual_subscriptions.rm (pipe_, send_unsubscription, this, false);
249 250 251
        //  Remove pipe without actually sending the message as it was taken
        //  care of by the manual call above. subscriptions is the real mtrie,
        //  so the pipe must be removed from there or it will be left over.
252
        _subscriptions.rm (pipe_, stub, (void *) NULL, false);
253
    } else {
254 255 256
        //  Remove the pipe from the trie. If there are topics that nobody
        //  is interested in anymore, send corresponding unsubscriptions
        //  upstream.
257
        _subscriptions.rm (pipe_, send_unsubscription, this, !_verbose_unsubs);
258
    }
259

260
    _dist.pipe_terminated (pipe_);
261 262
}

263
void zmq::xpub_t::mark_as_matching (pipe_t *pipe_, xpub_t *self_)
264
{
265
    self_->_dist.match (pipe_);
266 267
}

268
int zmq::xpub_t::xsend (msg_t *msg_)
269
{
270
    bool msg_more = (msg_->flags () & msg_t::more) != 0;
271 272

    //  For the first part of multi-part message, find the matching pipes.
273 274 275
    if (!_more) {
        _subscriptions.match (static_cast<unsigned char *> (msg_->data ()),
                              msg_->size (), mark_as_matching, this);
276 277
        // If inverted matching is used, reverse the selection now
        if (options.invert_matching) {
278
            _dist.reverse_match ();
279 280
        }
    }
281

282
    int rc = -1; //  Assume we fail
283 284
    if (_lossy || _dist.check_hwm ()) {
        if (_dist.send_to_matching (msg_) == 0) {
285
            //  If we are at the end of multi-part message we can mark
286 287
            //  all the pipes as non-matching.
            if (!msg_more)
288 289
                _dist.unmatch ();
            _more = msg_more;
290
            rc = 0; //  Yay, sent successfully
291
        }
292
    } else
293 294
        errno = EAGAIN;
    return rc;
295 296 297 298
}

bool zmq::xpub_t::xhas_out ()
{
299
    return _dist.has_out ();
300 301
}

302
int zmq::xpub_t::xrecv (msg_t *msg_)
303
{
Martin Hurton's avatar
Martin Hurton committed
304
    //  If there is at least one
305
    if (_pending_data.empty ()) {
306 307 308
        errno = EAGAIN;
        return -1;
    }
309

310
    // User is reading a message, set last_pipe and remove it from the deque
311 312 313
    if (_manual && !_pending_pipes.empty ()) {
        _last_pipe = _pending_pipes.front ();
        _pending_pipes.pop_front ();
314
    }
315

316 317
    int rc = msg_->close ();
    errno_assert (rc == 0);
318
    rc = msg_->init_size (_pending_data.front ().size ());
319
    errno_assert (rc == 0);
320 321
    memcpy (msg_->data (), _pending_data.front ().data (),
            _pending_data.front ().size ());
322 323

    // set metadata only if there is some
324
    if (metadata_t *metadata = _pending_metadata.front ()) {
325
        msg_->set_metadata (metadata);
326
        // Remove ref corresponding to vector placement
327
        metadata->drop_ref ();
328
    }
329

330 331 332 333
    msg_->set_flags (_pending_flags.front ());
    _pending_data.pop_front ();
    _pending_metadata.pop_front ();
    _pending_flags.pop_front ();
334
    return 0;
335 336 337 338
}

bool zmq::xpub_t::xhas_in ()
{
339
    return !_pending_data.empty ();
340 341
}

342
void zmq::xpub_t::send_unsubscription (zmq::mtrie_t::prefix_t data_,
343
                                       size_t size_,
344
                                       xpub_t *self_)
345
{
346
    if (self_->options.type != ZMQ_PUB) {
347 348
        //  Place the unsubscription to the queue of pending (un)subscriptions
        //  to be retrieved by the user later on.
349
        blob_t unsub (size_ + 1);
350
        *unsub.data () = 0;
Martin Hurton's avatar
Martin Hurton committed
351
        if (size_ > 0)
352
            memcpy (unsub.data () + 1, data_, size_);
353 354 355
        self_->_pending_data.ZMQ_PUSH_OR_EMPLACE_BACK (ZMQ_MOVE (unsub));
        self_->_pending_metadata.push_back (NULL);
        self_->_pending_flags.push_back (0);
356

357 358 359
        if (self_->_manual) {
            self_->_last_pipe = NULL;
            self_->_pending_pipes.push_back (NULL);
360
        }
361
    }
362
}