mirror of
https://github.com/zeromq/libzmq.git
synced 2024-12-13 02:42:58 +01:00
d70714e877
Solution: add some msg helpers to parse commands, and check for subscribe or cancel commands and process them accordingly in the xpub and xsub classes.
306 lines
9.3 KiB
C++
306 lines
9.3 KiB
C++
/*
|
|
Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file
|
|
|
|
This file is part of libzmq, the ZeroMQ core engine in C++.
|
|
|
|
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
|
|
(at your option) any later version.
|
|
|
|
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.
|
|
|
|
You should have received a copy of the GNU Lesser General Public License
|
|
along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
*/
|
|
|
|
#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).
|
|
|
|
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,
|
|
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_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;
|
|
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 ();
|
|
bool is_vsm () const;
|
|
bool is_cmsg () const;
|
|
bool is_lmsg () const;
|
|
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_);
|
|
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_);
|
|
|
|
// 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
|
|
};
|
|
|
|
// 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 + 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;
|
|
} _u;
|
|
};
|
|
|
|
inline int close_and_return (zmq::msg_t *msg_, int echo_)
|
|
{
|
|
// Since we abort on close failure we preserve errno for success case.
|
|
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
|