radio.cpp

/* SPDX-License-Identifier: MPL-2.0 */

#include "precompiled.hpp"
#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_) :
    socket_base_t (parent_, tid_, sid_, true), _lossy (true)
{
    options.type = ZMQ_RADIO;
}

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

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

    zmq_assert (pipe_);

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

    _dist.attach (pipe_);

    if (subscribe_to_all_)
        _udp_pipes.push_back (pipe_);
    //  The pipe is active when attached. Let's read the subscriptions from
    //  it, if any.
    else
        xread_activated (pipe_);
}

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

            if (msg.is_join ())
                _subscriptions.ZMQ_MAP_INSERT_OR_EMPLACE (ZMQ_MOVE (group),
                                                          pipe_);
            else {
                std::pair<subscriptions_t::iterator, subscriptions_t::iterator>
                  range = _subscriptions.equal_range (group);

                for (subscriptions_t::iterator it = range.first;
                     it != range.second; ++it) {
                    if (it->second == pipe_) {
                        _subscriptions.erase (it);
                        break;
                    }
                }
            }
        }
        msg.close ();
    }
}

void zmq::radio_t::xwrite_activated (pipe_t *pipe_)
{
    _dist.activated (pipe_);
}
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)
        _lossy = (*static_cast<const int *> (optval_) == 0);
    else {
        errno = EINVAL;
        return -1;
    }
    return 0;
}

void zmq::radio_t::xpipe_terminated (pipe_t *pipe_)
{
    for (subscriptions_t::iterator it = _subscriptions.begin (),
                                   end = _subscriptions.end ();
         it != end;) {
        if (it->second == pipe_) {
#if __cplusplus >= 201103L || (defined _MSC_VER && _MSC_VER >= 1700)
            it = _subscriptions.erase (it);
#else
            _subscriptions.erase (it++);
#endif
        } else {
            ++it;
        }
    }

    {
        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);
    }

    _dist.pipe_terminated (pipe_);
}

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;
    }

    _dist.unmatch ();

    const std::pair<subscriptions_t::iterator, subscriptions_t::iterator>
      range = _subscriptions.equal_range (std::string (msg_->group ()));

    for (subscriptions_t::iterator it = range.first; it != range.second; ++it)
        _dist.match (it->second);

    for (udp_pipes_t::iterator it = _udp_pipes.begin (),
                               end = _udp_pipes.end ();
         it != end; ++it)
        _dist.match (*it);

    int rc = -1;
    if (_lossy || _dist.check_hwm ()) {
        if (_dist.send_to_matching (msg_) == 0) {
            rc = 0; //  Yay, sent successfully
        }
    } else
        errno = EAGAIN;

    return rc;
}

bool zmq::radio_t::xhas_out ()
{
    return _dist.has_out ();
}

int zmq::radio_t::xrecv (msg_t *msg_)
{
    //  Messages cannot be received from PUB socket.
    LIBZMQ_UNUSED (msg_);
    errno = ENOTSUP;
    return -1;
}

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

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_) :
    session_base_t (io_thread_, connect_, socket_, options_, addr_),
    _state (group)
{
}

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

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

        int group_length;
        const char *group;

        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) {
            group_length = static_cast<int> (data_size) - 5;
            group = command_data + 5;
            rc = join_leave_msg.init_join ();
        } else if (data_size >= 6 && memcmp (command_data, "\5LEAVE", 6) == 0) {
            group_length = static_cast<int> (data_size) - 6;
            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_);
    }
    return session_base_t::push_msg (msg_);
}

int zmq::radio_session_t::pull_msg (msg_t *msg_)
{
    if (_state == group) {
        int rc = session_base_t::pull_msg (&_pending_msg);
        if (rc != 0)
            return rc;

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

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

        //  Next status is the body
        _state = body;
        return 0;
    }
    *msg_ = _pending_msg;
    _state = group;
    return 0;
}

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