/* SPDX-License-Identifier: MPL-2.0 */
#ifndef __ZMQ_MSG_HPP_INCLUDE__
#define __ZMQ_MSG_HPP_INCLUDE__
#include <stddef.h>
#include <stdio.h>
#include "config.hpp"
#include "err.hpp"
#include "fd.hpp"
#include "atomic_counter.hpp"
#include "metadata.hpp"
// bits 2-5
#define CMD_TYPE_MASK 0x1c
// 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.
extern "C" {
typedef void (msg_free_fn) (void *data_, void *hint_);
}
namespace zmq
{
// Note that this structure needs to be explicitly constructed
// (init functions) and destructed (close function).
static const char cancel_cmd_name[] = "\6CANCEL";
static const char sub_cmd_name[] = "\x9SUBSCRIBE";
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
{
void *data;
size_t size;
msg_free_fn *ffn;
void *hint;
zmq::atomic_counter_t refcnt;
};
// Message flags.
enum
{
more = 1, // Followed by more parts
command = 2, // Command frame (see ZMTP spec)
// 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,
close_cmd = 20,
credential = 32,
routing_id = 64,
shared = 128
};
bool check () const;
int init ();
int init (void *data_,
size_t size_,
msg_free_fn *ffn_,
void *hint_,
content_t *content_ = NULL);
int init_size (size_t size_);
int init_buffer (const void *buf_, 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 init_subscribe (const size_t size_, const unsigned char *topic);
int init_cancel (const size_t size_, const unsigned char *topic);
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;
bool is_ping () const;
bool is_pong () const;
bool is_close_cmd () 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
bool is_subscribe () const
{
return (_u.base.flags & CMD_TYPE_MASK) == subscribe;
}
bool is_cancel () const
{
return (_u.base.flags & CMD_TYPE_MASK) == cancel;
}
size_t command_body_size () const;
void *command_body ();
bool is_vsm () const;
bool is_cmsg () const;
bool is_lmsg () const;
bool is_zcmsg () const;
uint32_t get_routing_id () const;
int set_routing_id (uint32_t routing_id_);
int reset_routing_id ();
const char *group () const;
int set_group (const char *group_);
int set_group (const char *, size_t length_);
// After calling this function you can copy the message in POD-style
// refs_ times. No need to call copy.
void add_refs (int refs_);
// 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_);
void shrink (size_t new_size_);
// 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))
};
enum
{
ping_cmd_name_size = 5, // 4PING
cancel_cmd_name_size = 7, // 6CANCEL
sub_cmd_name_size = 10 // 9SUBSCRIBE
};
private:
zmq::atomic_counter_t *refcnt ();
// 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,
// zero-copy LMSG message for v2_decoder
type_zclmsg = 105,
// Join message for radio_dish
type_join = 106,
// Leave message for radio_dish
type_leave = 107,
type_max = 107
};
enum group_type_t
{
group_type_short,
group_type_long
};
struct long_group_t
{
char group[ZMQ_GROUP_MAX_LENGTH + 1];
atomic_counter_t refcnt;
};
union group_t
{
unsigned char type;
struct
{
unsigned char type;
char group[15];
} sgroup;
struct
{
unsigned char type;
long_group_t *content;
} lgroup;
};
// 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
{
struct
{
metadata_t *metadata;
unsigned char unused[msg_t_size
- (sizeof (metadata_t *) + 2
+ sizeof (uint32_t) + sizeof (group_t))];
unsigned char type;
unsigned char flags;
uint32_t routing_id;
group_t group;
} base;
struct
{
metadata_t *metadata;
unsigned char data[max_vsm_size];
unsigned char size;
unsigned char type;
unsigned char flags;
uint32_t routing_id;
group_t group;
} vsm;
struct
{
metadata_t *metadata;
content_t *content;
unsigned char
unused[msg_t_size
- (sizeof (metadata_t *) + sizeof (content_t *) + 2
+ sizeof (uint32_t) + sizeof (group_t))];
unsigned char type;
unsigned char flags;
uint32_t routing_id;
group_t group;
} lmsg;
struct
{
metadata_t *metadata;
content_t *content;
unsigned char
unused[msg_t_size
- (sizeof (metadata_t *) + sizeof (content_t *) + 2
+ sizeof (uint32_t) + sizeof (group_t))];
unsigned char type;
unsigned char flags;
uint32_t routing_id;
group_t group;
} 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 + sizeof (uint32_t)
+ sizeof (group_t))];
unsigned char type;
unsigned char flags;
uint32_t routing_id;
group_t group;
} cmsg;
struct
{
metadata_t *metadata;
unsigned char unused[msg_t_size
- (sizeof (metadata_t *) + 2
+ sizeof (uint32_t) + sizeof (group_t))];
unsigned char type;
unsigned char flags;
uint32_t routing_id;
group_t group;
} delimiter;
} _u;
};
inline int close_and_return (zmq::msg_t *msg_, int echo_)
{
// Since we abort on close failure we preserve errno for success case.
const int err = errno;
const int rc = msg_->close ();
errno_assert (rc == 0);
errno = err;
return echo_;
}
inline int close_and_return (zmq::msg_t msg_[], int count_, int echo_)
{
for (int i = 0; i < count_; i++)
close_and_return (&msg_[i], 0);
return echo_;
}
}
#endif
↑ V1071 Consider inspecting the 'close_and_return' function. The return value is not always used. Total calls: 28, discarded results: 1.