server.cpp 5.26 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
#include "macros.hpp"
32 33 34 35 36 37 38 39
#include "server.hpp"
#include "pipe.hpp"
#include "wire.hpp"
#include "random.hpp"
#include "likely.hpp"
#include "err.hpp"

zmq::server_t::server_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
somdoron's avatar
somdoron committed
40
    socket_base_t (parent_, tid_, sid_, true),
41
    _next_routing_id (generate_random ())
42
{
43
    options.type = ZMQ_SERVER;
44 45 46
}

zmq::server_t::~server_t ()
47
{
48
    zmq_assert (_out_pipes.empty ());
49 50
}

51 52 53
void zmq::server_t::xattach_pipe (pipe_t *pipe_,
                                  bool subscribe_to_all_,
                                  bool locally_initiated_)
54
{
55
    LIBZMQ_UNUSED (subscribe_to_all_);
56
    LIBZMQ_UNUSED (locally_initiated_);
57

58
    zmq_assert (pipe_);
59

60
    uint32_t routing_id = _next_routing_id++;
61
    if (!routing_id)
62
        routing_id = _next_routing_id++; //  Never use Routing ID zero
63

64
    pipe_->set_server_socket_routing_id (routing_id);
65 66
    //  Add the record into output pipes lookup table
    outpipe_t outpipe = {pipe_, true};
67
    bool ok = _out_pipes.ZMQ_MAP_INSERT_OR_EMPLACE (routing_id, outpipe).second;
68
    zmq_assert (ok);
69

70
    _fq.attach (pipe_);
71 72 73 74
}

void zmq::server_t::xpipe_terminated (pipe_t *pipe_)
{
75 76 77 78 79
    out_pipes_t::iterator it =
      _out_pipes.find (pipe_->get_server_socket_routing_id ());
    zmq_assert (it != _out_pipes.end ());
    _out_pipes.erase (it);
    _fq.pipe_terminated (pipe_);
80 81 82
}

void zmq::server_t::xread_activated (pipe_t *pipe_)
83
{
84
    _fq.activated (pipe_);
85 86 87 88
}

void zmq::server_t::xwrite_activated (pipe_t *pipe_)
{
89
    const out_pipes_t::iterator end = _out_pipes.end ();
90
    out_pipes_t::iterator it;
91
    for (it = _out_pipes.begin (); it != end; ++it)
92 93 94
        if (it->second.pipe == pipe_)
            break;

95
    zmq_assert (it != _out_pipes.end ());
96 97 98 99 100 101
    zmq_assert (!it->second.active);
    it->second.active = true;
}

int zmq::server_t::xsend (msg_t *msg_)
{
102 103 104 105 106
    //  SERVER sockets do not allow multipart data (ZMQ_SNDMORE)
    if (msg_->flags () & msg_t::more) {
        errno = EINVAL;
        return -1;
    }
107
    //  Find the pipe associated with the routing stored in the message.
108
    uint32_t routing_id = msg_->get_routing_id ();
109
    out_pipes_t::iterator it = _out_pipes.find (routing_id);
110

111
    if (it != _out_pipes.end ()) {
112
        if (!it->second.pipe->check_write ()) {
113
            it->second.active = false;
114
            errno = EAGAIN;
115
            return -1;
116
        }
117
    } else {
118 119 120
        errno = EHOSTUNREACH;
        return -1;
    }
121

Constantin Rack's avatar
Constantin Rack committed
122
    //  Message might be delivered over inproc, so we reset routing id
123 124 125
    int rc = msg_->reset_routing_id ();
    errno_assert (rc == 0);

126 127 128
    bool ok = it->second.pipe->write (msg_);
    if (unlikely (!ok)) {
        // Message failed to send - we must close it ourselves.
129
        rc = msg_->close ();
130
        errno_assert (rc == 0);
131
    } else
132 133
        it->second.pipe->flush ();

134
    //  Detach the message from the data buffer.
135
    rc = msg_->init ();
136 137 138 139 140 141
    errno_assert (rc == 0);

    return 0;
}

int zmq::server_t::xrecv (msg_t *msg_)
142
{
143
    pipe_t *pipe = NULL;
144
    int rc = _fq.recvpipe (msg_, &pipe);
145

146 147 148
    // Drop any messages with more flag
    while (rc == 0 && msg_->flags () & msg_t::more) {
        // drop all frames of the current multi-frame message
149
        rc = _fq.recvpipe (msg_, NULL);
150

151
        while (rc == 0 && msg_->flags () & msg_t::more)
152
            rc = _fq.recvpipe (msg_, NULL);
153

154
        // get the new message
155
        if (rc == 0)
156
            rc = _fq.recvpipe (msg_, &pipe);
157
    }
158

159
    if (rc != 0)
160
        return rc;
161 162

    zmq_assert (pipe != NULL);
163

164
    uint32_t routing_id = pipe->get_server_socket_routing_id ();
165
    msg_->set_routing_id (routing_id);
166 167 168 169 170 171

    return 0;
}

bool zmq::server_t::xhas_in ()
{
172
    return _fq.has_in ();
173 174 175 176 177
}

bool zmq::server_t::xhas_out ()
{
    //  In theory, SERVER socket is always ready for writing. Whether actual
178
    //  attempt to write succeeds depends on which pipe the message is going
179 180 181
    //  to be routed to.
    return true;
}