zmq_engine.cpp 4.72 KB
Newer Older
1
/*
2
    Copyright (c) 2007-2010 iMatix Corporation
3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19

    This file is part of 0MQ.

    0MQ is free software; you can redistribute it and/or modify it under
    the terms of the Lesser GNU General Public License as published by
    the Free Software Foundation; either version 3 of the License, or
    (at your option) any later version.

    0MQ 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
    Lesser GNU General Public License for more details.

    You should have received a copy of the Lesser GNU General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

20 21
#include <new>

22
#include "zmq_engine.hpp"
23
#include "zmq_connecter.hpp"
24
#include "io_thread.hpp"
25 26 27
#include "i_inout.hpp"
#include "config.hpp"
#include "err.hpp"
28

29
zmq::zmq_engine_t::zmq_engine_t (io_thread_t *parent_, fd_t fd_,
30 31
      const options_t &options_, bool reconnect_, 
      const char *protocol_, const char *address_) :
32
    io_object_t (parent_),
33
    inpos (NULL),
34
    insize (0),
35
    decoder (in_batch_size, NULL, 0),
36
    outpos (NULL),
37
    outsize (0),
38
    encoder (out_batch_size, false),
39
    inout (NULL),
40 41
    options (options_),
    reconnect (reconnect_)
42
{
43 44
    if (reconnect) {
        protocol = protocol_;
45
        address = address_;
46
    }
47

48
    //  Initialise the underlying socket.
49
    int rc = tcp_socket.open (fd_, options.sndbuf, options.rcvbuf);
50
    zmq_assert (rc == 0);
51 52 53 54
}

zmq::zmq_engine_t::~zmq_engine_t ()
{
55 56 57 58
}

void zmq::zmq_engine_t::plug (i_inout *inout_)
{
59 60
    zmq_assert (!inout);

61 62 63 64 65 66 67 68
    encoder.set_inout (inout_);
    decoder.set_inout (inout_);

    handle = add_fd (tcp_socket.get_fd ());
    set_pollin (handle);
    set_pollout (handle);

    inout = inout_;
69 70 71

    //  Flush all the data that may have been already received downstream.
    in_event ();
72 73 74 75 76
}

void zmq::zmq_engine_t::unplug ()
{
    rm_fd (handle);
77 78
    encoder.set_inout (NULL);
    decoder.set_inout (NULL);
79 80 81 82 83
    inout = NULL;
}

void zmq::zmq_engine_t::in_event ()
{
84 85
    bool disconnection = false;

86 87
    //  If there's no data to process in the buffer...
    if (!insize) {
88

89 90 91
        //  Retrieve the buffer and read as much data as possible.
        decoder.get_buffer (&inpos, &insize);
        insize = tcp_socket.read (inpos, insize);
92 93

        //  Check whether the peer has closed the connection.
94
        if (insize == (size_t) -1) {
95
            insize = 0;
96
            disconnection = true;
97 98 99 100
        }
    }

    //  Push the data to the decoder.
101 102 103 104 105
    size_t processed = decoder.process_buffer (inpos, insize);

    //  Adjust the buffer.
    inpos += processed;
    insize -= processed;
106

107
    //  Stop polling for input if we got stuck.
108
    if (processed < insize) {
109 110 111 112 113

        //  This may happen if queue limits are implemented or when
        //  init object reads all required information from the socket
        //  and rejects to read more data.
        reset_pollin (handle);
114
    }
115

Martin Sustrik's avatar
Martin Sustrik committed
116 117
    //  Flush all messages the decoder may have produced.
    inout->flush ();
118 119 120

    if (disconnection)
        error ();
121 122
}

123
void zmq::zmq_engine_t::out_event ()
124
{
125
    //  If write buffer is empty, try to read new data from the encoder.
126
    if (!outsize) {
Martin Sustrik's avatar
Martin Sustrik committed
127 128 129

        outpos = NULL;
        encoder.get_data (&outpos, &outsize);
130
        
131
        //  If there is no data to send, stop polling for output.
132
        if (outsize == 0) {
133
            reset_pollout (handle);
134 135
            return;
        }
136 137 138 139
    }

    //  If there are any data to write in write buffer, write as much as
    //  possible to the socket.
140
    int nbytes = tcp_socket.write (outpos, outsize);
141

142 143 144 145
    //  Handle problems with the connection.
    if (nbytes == -1) {
        error ();
        return;
146
    }
147 148 149

    outpos += nbytes;
    outsize -= nbytes;
150 151
}

Martin Sustrik's avatar
Martin Sustrik committed
152 153 154
void zmq::zmq_engine_t::revive ()
{
    set_pollout (handle);
155 156 157 158 159 160

    //  Speculative write: The assumption is that at the moment new message
    //  was sent by the user the socket is probably available for writing.
    //  Thus we try to write the data to socket avoiding polling for POLLOUT.
    //  Consequently, the latency should be better in request/reply scenarios.
    out_event ();
Martin Sustrik's avatar
Martin Sustrik committed
161 162
}

163 164
void zmq::zmq_engine_t::error ()
{
Martin Sustrik's avatar
Martin Sustrik committed
165
    zmq_assert (inout);
166 167 168 169 170 171

    zmq_connecter_t *reconnecter = NULL;
    if (reconnect) {

        //  Create a connecter object to attempt reconnect.
        //  Ask it to wait for a while before reconnecting.
172
        reconnecter = new (std::nothrow) zmq_connecter_t (
173
            inout->get_io_thread (), inout->get_owner (),
174
            options, inout->get_ordinal (), true);
175
        zmq_assert (reconnecter);
176
        reconnecter->set_address (protocol.c_str(), address.c_str ());
177 178 179
    }

    inout->detach (reconnecter);
Martin Sustrik's avatar
Martin Sustrik committed
180 181
    unplug ();
    delete this;
182
}