msg.hpp

/* 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