session_base.hpp 4.95 KB
Newer Older
1
/*
2
    Copyright (c) 2007-2014 Contributors as noted in the AUTHORS file
3 4 5 6

    This file is part of 0MQ.

    0MQ is free software; you can redistribute it and/or modify it under
7
    the terms of the GNU Lesser General Public License as published by
8 9 10 11 12 13
    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
14
    GNU Lesser General Public License for more details.
15

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

20 21
#ifndef __ZMQ_SESSION_BASE_HPP_INCLUDED__
#define __ZMQ_SESSION_BASE_HPP_INCLUDED__
22

23
#include <string>
24
#include <stdarg.h>
25

26
#include "own.hpp"
27
#include "io_object.hpp"
28
#include "pipe.hpp"
29
#include "socket_base.hpp"
30
#include "stream_engine.hpp"
31 32 33 34

namespace zmq
{

35 36 37 38
    class pipe_t;
    class io_thread_t;
    class socket_base_t;
    struct i_engine;
39
    struct address_t;
40

41
    class session_base_t :
42
        public own_t,
43
        public io_object_t,
44
        public i_pipe_events
45 46 47
    {
    public:

48
        //  Create a session of the particular type.
49
        static session_base_t *create (zmq::io_thread_t *io_thread_,
Martin Hurton's avatar
Martin Hurton committed
50
            bool active_, zmq::socket_base_t *socket_,
51
            const options_t &options_, address_t *addr_);
52

53
        //  To be used once only, when creating the session.
54
        void attach_pipe (zmq::pipe_t *pipe_);
55

56
        //  Following functions are the interface exposed towards the engine.
Martin Hurton's avatar
Martin Hurton committed
57
        virtual void reset ();
58
        void flush ();
59
        void engine_error (zmq::stream_engine_t::error_reason_t reason);
60

61
        //  i_pipe_events interface implementation.
62 63 64
        void read_activated (zmq::pipe_t *pipe_);
        void write_activated (zmq::pipe_t *pipe_);
        void hiccuped (zmq::pipe_t *pipe_);
65
        void pipe_terminated (zmq::pipe_t *pipe_);
Martin Sustrik's avatar
Martin Sustrik committed
66

67 68 69 70
        //  Delivers a message. Returns 0 if successful; -1 otherwise.
        //  The function takes ownership of the message.
        int push_msg (msg_t *msg_);

71 72
        int zap_connect ();

73 74 75 76 77
        //  Fetches a message. Returns 0 if successful; -1 otherwise.
        //  The caller is responsible for freeing the message when no
        //  longer used.
        int pull_msg (msg_t *msg_);

78 79 80 81 82 83 84 85 86 87
        //  Receives message from ZAP socket.
        //  Returns 0 on success; -1 otherwise.
        //  The caller is responsible for freeing the message.
        int read_zap_msg (msg_t *msg_);

        //  Sends message to ZAP socket.
        //  Returns 0 on success; -1 otherwise.
        //  The function takes ownership of the message.
        int write_zap_msg (msg_t *msg_);

88
        socket_base_t *get_socket ();
89

90
    protected:
91

Martin Hurton's avatar
Martin Hurton committed
92
        session_base_t (zmq::io_thread_t *io_thread_, bool active_,
93
            zmq::socket_base_t *socket_, const options_t &options_,
94
            address_t *addr_);
95
        virtual ~session_base_t ();
96 97

    private:
98

99
        void start_connecting (bool wait_);
100

Martin Hurton's avatar
Martin Hurton committed
101
        void reconnect ();
102

103 104
        //  Handlers for incoming commands.
        void process_plug ();
105
        void process_attach (zmq::i_engine *engine_);
106 107 108 109
        void process_term (int linger_);

        //  i_poll_events handlers.
        void timer_event (int id_);
110

111 112 113 114
        //  Remove any half processed messages. Flush unflushed messages.
        //  Call this function when engine disconnect to get rid of leftovers.
        void clean_pipes ();

115 116
        //  If true, this session (re)connects to the peer. Otherwise, it's
        //  a transient session created by the listener.
Martin Hurton's avatar
Martin Hurton committed
117
        const bool active;
118

119
        //  Pipe connecting the session to its socket.
120
        zmq::pipe_t *pipe;
Martin Hurton's avatar
Martin Hurton committed
121

122 123 124
        //  Pipe used to exchange messages with ZAP socket.
        zmq::pipe_t *zap_pipe;

125
        //  This set is added to with pipes we are disconnecting, but haven't yet completed
Martin Hurton's avatar
Martin Hurton committed
126
        std::set <pipe_t *> terminating_pipes;
127

128 129 130 131
        //  This flag is true if the remainder of the message being processed
        //  is still in the in pipe.
        bool incomplete_in;

132 133 134
        //  True if termination have been suspended to push the pending
        //  messages to the network.
        bool pending;
135

136
        //  The protocol I/O engine connected to the session.
137
        zmq::i_engine *engine;
138

139
        //  The socket the session belongs to.
140
        zmq::socket_base_t *socket;
141 142 143

        //  I/O thread the session is living in. It will be used to plug in
        //  the engines into the same thread.
144
        zmq::io_thread_t *io_thread;
145

146 147 148 149 150 151
        //  ID of the linger timer
        enum {linger_timer_id = 0x20};

        //  True is linger timer is running.
        bool has_linger_timer;

152
        //  Protocol and address to use when connecting.
153
        address_t *addr;
154

155 156
        session_base_t (const session_base_t&);
        const session_base_t &operator = (const session_base_t&);
157 158 159 160 161
    };

}

#endif