dish.cpp 8.82 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 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92
#include <string.h>

#include "macros.hpp"
#include "dish.hpp"
#include "err.hpp"

zmq::dish_t::dish_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
    socket_base_t (parent_, tid_, sid_, true),
    has_message (false)
{
    options.type = ZMQ_DISH;

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

    int rc = message.init ();
    errno_assert (rc == 0);
}

zmq::dish_t::~dish_t ()
{
    int rc = message.close ();
    errno_assert (rc == 0);
}

void zmq::dish_t::xattach_pipe (pipe_t *pipe_, bool subscribe_to_all_)
{
    LIBZMQ_UNUSED (subscribe_to_all_);

    zmq_assert (pipe_);
    fq.attach (pipe_);
    dist.attach (pipe_);

    //  Send all the cached subscriptions to the new upstream peer.
    send_subscriptions (pipe_);
}

void zmq::dish_t::xread_activated (pipe_t *pipe_)
{
    fq.activated (pipe_);
}

void zmq::dish_t::xwrite_activated (pipe_t *pipe_)
{
    dist.activated (pipe_);
}

void zmq::dish_t::xpipe_terminated (pipe_t *pipe_)
{
    fq.pipe_terminated (pipe_);
    dist.pipe_terminated (pipe_);
}

void zmq::dish_t::xhiccuped (pipe_t *pipe_)
{
    //  Send all the cached subscriptions to the hiccuped pipe.
    send_subscriptions (pipe_);
}

int zmq::dish_t::xjoin (const char* group_)
{
93 94 95
    std::string group = std::string (group_);

    if (group.length () > ZMQ_GROUP_MAX_LENGTH) {
somdoron's avatar
somdoron committed
96 97 98 99
        errno = EINVAL;
        return -1;
    }

100
    subscriptions_t::iterator it = subscriptions.find (group);
somdoron's avatar
somdoron committed
101 102 103 104 105 106 107

    //  User cannot join same group twice
    if (it != subscriptions.end ()) {
        errno = EINVAL;
        return -1;
    }

108
    subscriptions.insert (group);
somdoron's avatar
somdoron committed
109 110

    msg_t msg;
111
    int rc = msg.init_join ();
somdoron's avatar
somdoron committed
112 113
    errno_assert (rc == 0);

114 115
    rc = msg.set_group (group_);
    errno_assert (rc == 0);
somdoron's avatar
somdoron committed
116 117 118 119 120 121 122 123 124 125 126 127 128 129

    int err = 0;
    rc = dist.send_to_all (&msg);
    if (rc != 0)
        err = errno;
    int rc2 = msg.close ();
    errno_assert (rc2 == 0);
    if (rc != 0)
        errno = err;
    return rc;
}

int zmq::dish_t::xleave (const char* group_)
{
130 131 132
    std::string group = std::string (group_);

    if (group.length () > ZMQ_GROUP_MAX_LENGTH) {
somdoron's avatar
somdoron committed
133 134 135 136
        errno = EINVAL;
        return -1;
    }

137
    subscriptions_t::iterator it =  std::find (subscriptions.begin (), subscriptions.end (), group);
somdoron's avatar
somdoron committed
138 139 140 141 142 143 144 145 146

    if (it == subscriptions.end ()) {
        errno = EINVAL;
        return -1;
    }

    subscriptions.erase (it);

    msg_t msg;
147
    int rc = msg.init_leave ();
somdoron's avatar
somdoron committed
148 149
    errno_assert (rc == 0);

150 151
    rc = msg.set_group (group_);
    errno_assert (rc == 0);
somdoron's avatar
somdoron committed
152 153 154 155 156 157 158 159 160 161 162 163 164 165

    int err = 0;
    rc = dist.send_to_all (&msg);
    if (rc != 0)
        err = errno;
    int rc2 = msg.close ();
    errno_assert (rc2 == 0);
    if (rc != 0)
        errno = err;
    return rc;
}

int zmq::dish_t::xsend (msg_t *msg_)
{
166
    LIBZMQ_UNUSED (msg_);
somdoron's avatar
somdoron committed
167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187
    errno = ENOTSUP;
    return -1;
}

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

int zmq::dish_t::xrecv (msg_t *msg_)
{
    //  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;
        return 0;
    }

188
    while (true) {
somdoron's avatar
somdoron committed
189

190 191
        //  Get a message using fair queueing algorithm.
        int rc = fq.recv (msg_);
somdoron's avatar
somdoron committed
192

193 194 195 196 197 198 199 200 201 202
        //  If there's no message available, return immediately.
        //  The same when error occurs.
        if (rc != 0)
            return -1;

        //  Filtering non matching messages
        subscriptions_t::iterator it = subscriptions.find (std::string(msg_->group ()));
        if (it != subscriptions.end ())
            return 0;
    }
somdoron's avatar
somdoron committed
203 204 205 206 207 208 209 210 211
}

bool zmq::dish_t::xhas_in ()
{
    //  If there's already a message prepared by a previous call to zmq_poll,
    //  return straight ahead.
    if (has_message)
        return true;

212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228
    while (true) {
        //  Get a message using fair queueing algorithm.
        int rc = fq.recv (&message);

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

        //  Filtering non matching messages
        subscriptions_t::iterator it = subscriptions.find (std::string(message.group ()));
        if (it != subscriptions.end ()) {
            has_message = true;
            return true;
        }
somdoron's avatar
somdoron committed
229 230 231 232 233 234 235 236 237 238 239 240
    }
}

zmq::blob_t zmq::dish_t::get_credential () const
{
    return fq.get_credential ();
}

void zmq::dish_t::send_subscriptions (pipe_t *pipe_)
{
    for (subscriptions_t::iterator it = subscriptions.begin (); it != subscriptions.end (); ++it) {
        msg_t msg;
241 242 243 244
        int rc = msg.init_join ();
        errno_assert (rc == 0);

        rc = msg.set_group (it->c_str());
somdoron's avatar
somdoron committed
245 246 247
        errno_assert (rc == 0);

        //  Send it to the pipe.
248 249
        pipe_->write (&msg);
        msg.close ();
somdoron's avatar
somdoron committed
250 251 252 253
    }

    pipe_->flush ();
}
254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311

zmq::dish_session_t::dish_session_t (io_thread_t *io_thread_, bool connect_,
      socket_base_t *socket_, const options_t &options_,
      address_t *addr_) :
    session_base_t (io_thread_, connect_, socket_, options_, addr_),
    state (group)
{
}

zmq::dish_session_t::~dish_session_t ()
{
}

int zmq::dish_session_t::push_msg (msg_t *msg_)
{
    if (state == group) {
        if ((msg_->flags() & msg_t::more) != msg_t::more) {
            errno = EFAULT;
            return -1;
        }

        if (msg_->size() > ZMQ_GROUP_MAX_LENGTH) {
            errno = EFAULT;
            return -1;
        }

        group_msg = *msg_;
        state = body;

        int rc = msg_->init ();
        errno_assert (rc == 0);
        return 0;
    }
    else {
        //  Set the message group
        int rc = msg_->set_group ((char*)group_msg.data (), group_msg. size());
        errno_assert (rc == 0);

        //  We set the group, so we don't need the group_msg anymore
        rc = group_msg.close ();
        errno_assert (rc == 0);

        //  Thread safe socket doesn't support multipart messages
        if ((msg_->flags() & msg_t::more) == msg_t::more) {
            errno = EFAULT;
            return -1;
        }

        //  Push message to dish socket
        rc = session_base_t::push_msg (msg_);

        if (rc == 0)
            state = group;

        return rc;
    }
}

312 313 314 315 316 317 318 319 320 321
int zmq::dish_session_t::pull_msg (msg_t *msg_)
{
    int rc = session_base_t::pull_msg (msg_);

    if (rc != 0)
        return rc;

    if (!msg_->is_join () && !msg_->is_leave ())
        return rc;
    else {
322
        int group_length = (int) strlen (msg_->group ());
323 324 325 326 327

        msg_t command;
        int offset;

        if (msg_->is_join ()) {
328
            rc = command.init_size (group_length + 5);
329 330
            errno_assert(rc == 0);
            offset = 5;
331 332 333
            memcpy (command.data (), "\4JOIN", 5);
        }
        else {
334
            rc = command.init_size (group_length + 6);
335 336
            errno_assert(rc == 0);
            offset = 6;
337 338 339 340 341 342 343 344 345 346
            memcpy (command.data (), "\5LEAVE", 6);
        }

        command.set_flags (msg_t::command);
        char* command_data = (char*)command.data ();

        //  Copy the group
        memcpy (command_data + offset, msg_->group (), group_length);

        //  Close the join message
347
        rc = msg_->close ();
348 349 350 351 352 353 354 355
        errno_assert (rc == 0);

        *msg_ = command;

        return 0;
    }
}

356 357 358 359 360
void zmq::dish_session_t::reset ()
{
    session_base_t::reset ();
    state = group;
}