session_base.hpp 4.98 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
        int zap_connect ();
Min RK's avatar
Min RK committed
72
        bool zap_enabled ();
73

74 75 76 77 78
        //  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_);

79 80 81 82 83 84 85 86 87 88
        //  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_);

89
        socket_base_t *get_socket ();
90

91
    protected:
92

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

    private:
99

100
        void start_connecting (bool wait_);
101

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

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

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

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

116 117
        //  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
118
        const bool active;
119

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

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

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

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

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

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

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

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

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

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

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

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

}

#endif