msg.hpp 9.33 KB
Newer Older
Martin Sustrik's avatar
Martin Sustrik committed
1
/*
2
    Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file
Martin Sustrik's avatar
Martin Sustrik committed
3

4
    This file is part of libzmq, the ZeroMQ core engine in C++.
Martin Sustrik's avatar
Martin Sustrik committed
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
Martin Sustrik's avatar
Martin Sustrik committed
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.
Martin Sustrik's avatar
Martin Sustrik committed
25

26
    You should have received a copy of the GNU Lesser General Public License
Martin Sustrik's avatar
Martin Sustrik committed
27 28 29
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

Martin Sustrik's avatar
Martin Sustrik committed
30 31
#ifndef __ZMQ_MSG_HPP_INCLUDE__
#define __ZMQ_MSG_HPP_INCLUDE__
Martin Sustrik's avatar
Martin Sustrik committed
32 33

#include <stddef.h>
Pieter Hintjens's avatar
Pieter Hintjens committed
34
#include <stdio.h>
Martin Sustrik's avatar
Martin Sustrik committed
35

36
#include "config.hpp"
37
#include "err.hpp"
38
#include "fd.hpp"
Martin Sustrik's avatar
Martin Sustrik committed
39
#include "atomic_counter.hpp"
40
#include "metadata.hpp"
Martin Sustrik's avatar
Martin Sustrik committed
41

42 43 44
//  bits 2-5
#define CMD_TYPE_MASK 0x1c

45 46 47
//  Signature for free function to deallocate the message content.
//  Note that it has to be declared as "C" so that it is the same as
//  zmq_free_fn defined in zmq.h.
48
extern "C" {
49
typedef void(msg_free_fn) (void *data_, void *hint_);
50 51
}

52 53
namespace zmq
{
54 55
//  Note that this structure needs to be explicitly constructed
//  (init functions) and destructed (close function).
Martin Sustrik's avatar
Martin Sustrik committed
56

57 58 59 60 61 62 63 64 65 66 67
class msg_t
{
  public:
    //  Shared message buffer. Message data are either allocated in one
    //  continuous block along with this structure - thus avoiding one
    //  malloc/free pair or they are stored in user-supplied memory.
    //  In the latter case, ffn member stores pointer to the function to be
    //  used to deallocate the data. If the buffer is actually shared (there
    //  are at least 2 references to it) refcount member contains number of
    //  references.
    struct content_t
Martin Sustrik's avatar
Martin Sustrik committed
68
    {
69 70 71 72 73 74
        void *data;
        size_t size;
        msg_free_fn *ffn;
        void *hint;
        zmq::atomic_counter_t refcnt;
    };
75

76 77 78 79 80
    //  Message flags.
    enum
    {
        more = 1,    //  Followed by more parts
        command = 2, //  Command frame (see ZMTP spec)
81 82 83 84 85 86
        //  Command types, use only bits 2-5 and compare with ==, not bitwise,
        //  a command can never be of more that one type at the same time
        ping = 4,
        pong = 8,
        subscribe = 12,
        cancel = 16,
87 88 89 90
        credential = 32,
        routing_id = 64,
        shared = 128
    };
91

92 93
    bool check () const;
    int init ();
94

95
    int init (void *data_,
96 97
              size_t size_,
              msg_free_fn *ffn_,
98
              void *hint_,
99
              content_t *content_ = NULL);
100

101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126
    int init_size (size_t size_);
    int init_data (void *data_, size_t size_, msg_free_fn *ffn_, void *hint_);
    int init_external_storage (content_t *content_,
                               void *data_,
                               size_t size_,
                               msg_free_fn *ffn_,
                               void *hint_);
    int init_delimiter ();
    int init_join ();
    int init_leave ();
    int close ();
    int move (msg_t &src_);
    int copy (msg_t &src_);
    void *data ();
    size_t size () const;
    unsigned char flags () const;
    void set_flags (unsigned char flags_);
    void reset_flags (unsigned char flags_);
    metadata_t *metadata () const;
    void set_metadata (metadata_t *metadata_);
    void reset_metadata ();
    bool is_routing_id () const;
    bool is_credential () const;
    bool is_delimiter () const;
    bool is_join () const;
    bool is_leave () const;
127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142
    bool is_ping () const;
    bool is_pong () const;

    //  These are called on each message received by the session_base class,
    //  so get them inlined to avoid the overhead of 2 function calls per msg
    inline bool is_subscribe () const
    {
        return (_u.base.flags & CMD_TYPE_MASK) == subscribe;
    }
    inline bool is_cancel () const
    {
        return (_u.base.flags & CMD_TYPE_MASK) == cancel;
    }

    size_t command_body_size () const;
    void *command_body ();
143 144
    bool is_vsm () const;
    bool is_cmsg () const;
145
    bool is_lmsg () const;
146 147 148 149 150 151
    bool is_zcmsg () const;
    uint32_t get_routing_id ();
    int set_routing_id (uint32_t routing_id_);
    int reset_routing_id ();
    const char *group ();
    int set_group (const char *group_);
152
    int set_group (const char *, size_t length_);
153

154 155 156
    //  After calling this function you can copy the message in POD-style
    //  refs_ times. No need to call copy.
    void add_refs (int refs_);
157

158 159 160
    //  Removes references previously added by add_refs. If the number of
    //  references drops to 0, the message is closed and false is returned.
    bool rm_refs (int refs_);
161

162 163 164 165 166 167 168 169 170 171 172
    //  Size in bytes of the largest message that is still copied around
    //  rather than being reference-counted.
    enum
    {
        msg_t_size = 64
    };
    enum
    {
        max_vsm_size =
          msg_t_size - (sizeof (metadata_t *) + 3 + 16 + sizeof (uint32_t))
    };
173 174 175 176 177 178
    enum
    {
        ping_cmd_name_size = 5,   // 4PING
        cancel_cmd_name_size = 7, // 6CANCEL
        sub_cmd_name_size = 10    // 9SUBSCRIBE
    };
179

180 181
  private:
    zmq::atomic_counter_t *refcnt ();
182

183 184 185 186 187 188 189 190 191 192 193 194
    //  Different message types.
    enum type_t
    {
        type_min = 101,
        //  VSM messages store the content in the message itself
        type_vsm = 101,
        //  LMSG messages store the content in malloc-ed memory
        type_lmsg = 102,
        //  Delimiter messages are used in envelopes
        type_delimiter = 103,
        //  CMSG messages point to constant data
        type_cmsg = 104,
195

196 197
        // zero-copy LMSG message for v2_decoder
        type_zclmsg = 105,
198

199 200
        //  Join message for radio_dish
        type_join = 106,
201

202 203
        //  Leave message for radio_dish
        type_leave = 107,
204

205
        type_max = 107
Martin Sustrik's avatar
Martin Sustrik committed
206 207
    };

208 209 210 211 212
    //  Note that fields shared between different message types are not
    //  moved to the parent class (msg_t). This way we get tighter packing
    //  of the data. Shared fields can be accessed via 'base' member of
    //  the union.
    union
213
    {
214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283
        struct
        {
            metadata_t *metadata;
            unsigned char
              unused[msg_t_size
                     - (sizeof (metadata_t *) + 2 + 16 + sizeof (uint32_t))];
            unsigned char type;
            unsigned char flags;
            char group[16];
            uint32_t routing_id;
        } base;
        struct
        {
            metadata_t *metadata;
            unsigned char data[max_vsm_size];
            unsigned char size;
            unsigned char type;
            unsigned char flags;
            char group[16];
            uint32_t routing_id;
        } vsm;
        struct
        {
            metadata_t *metadata;
            content_t *content;
            unsigned char unused[msg_t_size
                                 - (sizeof (metadata_t *) + sizeof (content_t *)
                                    + 2 + 16 + sizeof (uint32_t))];
            unsigned char type;
            unsigned char flags;
            char group[16];
            uint32_t routing_id;
        } lmsg;
        struct
        {
            metadata_t *metadata;
            content_t *content;
            unsigned char unused[msg_t_size
                                 - (sizeof (metadata_t *) + sizeof (content_t *)
                                    + 2 + 16 + sizeof (uint32_t))];
            unsigned char type;
            unsigned char flags;
            char group[16];
            uint32_t routing_id;
        } zclmsg;
        struct
        {
            metadata_t *metadata;
            void *data;
            size_t size;
            unsigned char
              unused[msg_t_size
                     - (sizeof (metadata_t *) + sizeof (void *)
                        + sizeof (size_t) + 2 + 16 + sizeof (uint32_t))];
            unsigned char type;
            unsigned char flags;
            char group[16];
            uint32_t routing_id;
        } cmsg;
        struct
        {
            metadata_t *metadata;
            unsigned char
              unused[msg_t_size
                     - (sizeof (metadata_t *) + 2 + 16 + sizeof (uint32_t))];
            unsigned char type;
            unsigned char flags;
            char group[16];
            uint32_t routing_id;
        } delimiter;
284
    } _u;
285
};
286

287
inline int close_and_return (zmq::msg_t *msg_, int echo_)
288 289 290
{
    // Since we abort on close failure we preserve errno for success case.
    int err = errno;
291
    const int rc = msg_->close ();
292 293
    errno_assert (rc == 0);
    errno = err;
294
    return echo_;
295 296
}

297
inline int close_and_return (zmq::msg_t msg_[], int count_, int echo_)
298
{
299 300 301
    for (int i = 0; i < count_; i++)
        close_and_return (&msg_[i], 0);
    return echo_;
302
}
303
}
Martin Sustrik's avatar
Martin Sustrik committed
304 305

#endif