radio.cpp 8.11 KB
Newer Older
somdoron's avatar
somdoron committed
1
/*
2
    Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file
somdoron's avatar
somdoron committed
3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29

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

    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
    (at your option) any later version.

    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.

    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"
somdoron's avatar
somdoron committed
31 32 33 34 35 36 37 38 39
#include <string.h>

#include "radio.hpp"
#include "macros.hpp"
#include "pipe.hpp"
#include "err.hpp"
#include "msg.hpp"

zmq::radio_t::radio_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
40
    socket_base_t (parent_, tid_, sid_, true),
41
    _lossy (true)
somdoron's avatar
somdoron committed
42 43 44 45 46 47 48 49
{
    options.type = ZMQ_RADIO;
}

zmq::radio_t::~radio_t ()
{
}

50 51 52
void zmq::radio_t::xattach_pipe (pipe_t *pipe_,
                                 bool subscribe_to_all_,
                                 bool locally_initiated_)
somdoron's avatar
somdoron committed
53 54
{
    LIBZMQ_UNUSED (subscribe_to_all_);
55
    LIBZMQ_UNUSED (locally_initiated_);
somdoron's avatar
somdoron committed
56 57 58 59 60 61 62

    zmq_assert (pipe_);

    //  Don't delay pipe termination as there is no one
    //  to receive the delimiter.
    pipe_->set_nodelay ();

63
    _dist.attach (pipe_);
somdoron's avatar
somdoron committed
64

65
    if (subscribe_to_all_)
66
        _udp_pipes.push_back (pipe_);
somdoron's avatar
somdoron committed
67 68
    //  The pipe is active when attached. Let's read the subscriptions from
    //  it, if any.
69 70
    else
        xread_activated (pipe_);
somdoron's avatar
somdoron committed
71 72 73 74 75
}

void zmq::radio_t::xread_activated (pipe_t *pipe_)
{
    //  There are some subscriptions waiting. Let's process them.
76 77
    msg_t msg;
    while (pipe_->read (&msg)) {
somdoron's avatar
somdoron committed
78
        //  Apply the subscription to the trie
79 80
        if (msg.is_join () || msg.is_leave ()) {
            std::string group = std::string (msg.group ());
somdoron's avatar
somdoron committed
81

82
            if (msg.is_join ())
83 84
                _subscriptions.ZMQ_MAP_INSERT_OR_EMPLACE (ZMQ_MOVE (group),
                                                          pipe_);
somdoron's avatar
somdoron committed
85
            else {
86
                std::pair<subscriptions_t::iterator, subscriptions_t::iterator>
87
                  range = _subscriptions.equal_range (group);
somdoron's avatar
somdoron committed
88

89 90
                for (subscriptions_t::iterator it = range.first;
                     it != range.second; ++it) {
somdoron's avatar
somdoron committed
91
                    if (it->second == pipe_) {
92
                        _subscriptions.erase (it);
somdoron's avatar
somdoron committed
93 94 95 96 97
                        break;
                    }
                }
            }
        }
98
        msg.close ();
somdoron's avatar
somdoron committed
99 100 101 102 103
    }
}

void zmq::radio_t::xwrite_activated (pipe_t *pipe_)
{
104
    _dist.activated (pipe_);
somdoron's avatar
somdoron committed
105
}
106 107 108 109 110 111 112 113 114
int zmq::radio_t::xsetsockopt (int option_,
                               const void *optval_,
                               size_t optvallen_)
{
    if (optvallen_ != sizeof (int) || *static_cast<const int *> (optval_) < 0) {
        errno = EINVAL;
        return -1;
    }
    if (option_ == ZMQ_XPUB_NODROP)
115
        _lossy = (*static_cast<const int *> (optval_) == 0);
116 117 118 119 120 121
    else {
        errno = EINVAL;
        return -1;
    }
    return 0;
}
somdoron's avatar
somdoron committed
122 123 124

void zmq::radio_t::xpipe_terminated (pipe_t *pipe_)
{
125 126 127
    for (subscriptions_t::iterator it = _subscriptions.begin (),
                                   end = _subscriptions.end ();
         it != end;) {
somdoron's avatar
somdoron committed
128
        if (it->second == pipe_) {
129 130 131
#if __cplusplus >= 201103L
            it = _subscriptions.erase (it);
#else
132
            _subscriptions.erase (it++);
133
#endif
134 135
        } else {
            ++it;
somdoron's avatar
somdoron committed
136 137 138
        }
    }

139 140 141 142 143 144 145
    {
        const udp_pipes_t::iterator end = _udp_pipes.end ();
        const udp_pipes_t::iterator it =
          std::find (_udp_pipes.begin (), end, pipe_);
        if (it != end)
            _udp_pipes.erase (it);
    }
146

147
    _dist.pipe_terminated (pipe_);
somdoron's avatar
somdoron committed
148 149 150 151 152 153 154 155 156 157
}

int zmq::radio_t::xsend (msg_t *msg_)
{
    //  Radio sockets do not allow multipart data (ZMQ_SNDMORE)
    if (msg_->flags () & msg_t::more) {
        errno = EINVAL;
        return -1;
    }

158
    _dist.unmatch ();
somdoron's avatar
somdoron committed
159

160 161
    const std::pair<subscriptions_t::iterator, subscriptions_t::iterator>
      range = _subscriptions.equal_range (std::string (msg_->group ()));
somdoron's avatar
somdoron committed
162

163
    for (subscriptions_t::iterator it = range.first; it != range.second; ++it)
164
        _dist.match (it->second);
somdoron's avatar
somdoron committed
165

166 167 168
    for (udp_pipes_t::iterator it = _udp_pipes.begin (),
                               end = _udp_pipes.end ();
         it != end; ++it)
169
        _dist.match (*it);
170

171
    int rc = -1;
172 173
    if (_lossy || _dist.check_hwm ()) {
        if (_dist.send_to_matching (msg_) == 0) {
174 175 176 177
            rc = 0; //  Yay, sent successfully
        }
    } else
        errno = EAGAIN;
somdoron's avatar
somdoron committed
178 179 180 181 182 183

    return rc;
}

bool zmq::radio_t::xhas_out ()
{
184
    return _dist.has_out ();
somdoron's avatar
somdoron committed
185 186 187 188 189
}

int zmq::radio_t::xrecv (msg_t *msg_)
{
    //  Messages cannot be received from PUB socket.
190
    LIBZMQ_UNUSED (msg_);
somdoron's avatar
somdoron committed
191 192 193 194 195 196 197 198
    errno = ENOTSUP;
    return -1;
}

bool zmq::radio_t::xhas_in ()
{
    return false;
}
199

200 201 202 203 204
zmq::radio_session_t::radio_session_t (io_thread_t *io_thread_,
                                       bool connect_,
                                       socket_base_t *socket_,
                                       const options_t &options_,
                                       address_t *addr_) :
205
    session_base_t (io_thread_, connect_, socket_, options_, addr_),
206
    _state (group)
207 208 209 210 211 212 213
{
}

zmq::radio_session_t::~radio_session_t ()
{
}

214 215
int zmq::radio_session_t::push_msg (msg_t *msg_)
{
216 217
    if (msg_->flags () & msg_t::command) {
        char *command_data = static_cast<char *> (msg_->data ());
218 219 220
        const size_t data_size = msg_->size ();

        int group_length;
221
        const char *group;
222 223 224 225 226 227

        msg_t join_leave_msg;
        int rc;

        //  Set the msg type to either JOIN or LEAVE
        if (data_size >= 5 && memcmp (command_data, "\4JOIN", 5) == 0) {
228
            group_length = static_cast<int> (data_size) - 5;
229 230
            group = command_data + 5;
            rc = join_leave_msg.init_join ();
231
        } else if (data_size >= 6 && memcmp (command_data, "\5LEAVE", 6) == 0) {
232
            group_length = static_cast<int> (data_size) - 6;
233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252
            group = command_data + 6;
            rc = join_leave_msg.init_leave ();
        }
        //  If it is not a JOIN or LEAVE just push the message
        else
            return session_base_t::push_msg (msg_);

        errno_assert (rc == 0);

        //  Set the group
        rc = join_leave_msg.set_group (group, group_length);
        errno_assert (rc == 0);

        //  Close the current command
        rc = msg_->close ();
        errno_assert (rc == 0);

        //  Push the join or leave command
        *msg_ = join_leave_msg;
        return session_base_t::push_msg (msg_);
253 254
    }
    return session_base_t::push_msg (msg_);
255 256
}

257 258
int zmq::radio_session_t::pull_msg (msg_t *msg_)
{
259 260
    if (_state == group) {
        int rc = session_base_t::pull_msg (&_pending_msg);
261 262 263
        if (rc != 0)
            return rc;

264
        const char *group = _pending_msg.group ();
265
        const int length = static_cast<int> (strlen (group));
266 267

        //  First frame is the group
268
        rc = msg_->init_size (length);
269 270
        errno_assert (rc == 0);
        msg_->set_flags (msg_t::more);
271 272 273
        memcpy (msg_->data (), group, length);

        //  Next status is the body
274
        _state = body;
275 276
        return 0;
    }
277 278
    *msg_ = _pending_msg;
    _state = group;
279
    return 0;
280 281 282 283 284
}

void zmq::radio_session_t::reset ()
{
    session_base_t::reset ();
285
    _state = group;
286
}