| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123 | // Copyright Takatoshi Kondo 2018//// Distributed under the Boost Software License, Version 1.0.// (See accompanying file LICENSE_1_0.txt or copy at// http://www.boost.org/LICENSE_1_0.txt)#if !defined(MQTT_MESSAGE_HPP)#define MQTT_MESSAGE_HPP#include <string>#include <vector>#include <memory>#include <algorithm>#include <numeric>#include <boost/asio/buffer.hpp>#include <boost/container/static_vector.hpp>#include <boost/numeric/conversion/cast.hpp>#include <mqtt/namespace.hpp>#include <mqtt/two_byte_util.hpp>#include <mqtt/fixed_header.hpp>#include <mqtt/remaining_length.hpp>#include <mqtt/subscribe_options.hpp>#include <mqtt/const_buffer_util.hpp>#include <mqtt/will.hpp>#include <mqtt/connect_flags.hpp>#include <mqtt/publish.hpp>#include <mqtt/exception.hpp>#include <mqtt/utf8encoded_strings.hpp>#include <mqtt/four_byte_util.hpp>#include <mqtt/packet_id_type.hpp>#include <mqtt/optional.hpp>#include <mqtt/string_view.hpp>#include <mqtt/property.hpp>#include <mqtt/string_check.hpp>#include <mqtt/move.hpp>#include <mqtt/reason_code.hpp>#include <mqtt/connect_return_code.hpp>#include <mqtt/publish.hpp>namespace MQTT_NS {namespace as = boost::asio;inline namespace v3_1_1 {namespace detail_v3_1_1 {class header_only_message {public:    /**     * @brief Create empty header_packet_id_message.     */    header_only_message(control_packet_type type, std::uint8_t flags)        : message_ { static_cast<char>(make_fixed_header(type, flags)), 0 }    {}    /**     * @brief Create const buffer sequence     *        it is for boost asio APIs     * @return const buffer sequence     */    std::vector<as::const_buffer> const_buffer_sequence() const {        return { as::buffer(message_.data(), message_.size()) };    }    /**     * @brief Get whole size of sequence     * @return whole size     */    std::size_t size() const {        return message_.size();    }    /**     * @brief Get number of element of const_buffer_sequence     * @return number of element of const_buffer_sequence     */    static constexpr std::size_t num_of_const_buffer_sequence() {        return 1;    }    /**     * @brief Create one continuours buffer.     *        All sequence of buffers are concatinated.     *        It is useful to store to file/database.     * @return continuous buffer     */    std::string continuous_buffer() const {        return std::string(message_.data(), size());    }private:    boost::container::static_vector<char, 2> message_;};template <std::size_t PacketIdBytes>class basic_header_packet_id_message;template <std::size_t PacketIdBytes>class basic_header_packet_id_message {public:    /**     * @brief Create empty header_packet_id_message.     */    basic_header_packet_id_message(control_packet_type type, std::uint8_t flags, typename packet_id_type<PacketIdBytes>::type packet_id)        : message_ { static_cast<char>(make_fixed_header(type, flags)), PacketIdBytes }    {        add_packet_id_to_buf<PacketIdBytes>::apply(message_, packet_id);    }    template <typename Iterator>    basic_header_packet_id_message(Iterator b, Iterator e) {        if (std::distance(b, e) != 2 + PacketIdBytes) throw remaining_length_error();        if (b[1] != PacketIdBytes) throw remaining_length_error();        std::copy(b, e, std::back_inserter(message_));    }    /**     * @brief Create const buffer sequence     *        it is for boost asio APIs     * @return const buffer sequence     */    std::vector<as::const_buffer> const_buffer_sequence() const {        return { as::buffer(message_.data(), size()) };    }    /**     * @brief Get whole size of sequence     * @return whole size     */    std::size_t size() const {        return message_.size();    }    /**     * @brief Get number of element of const_buffer_sequence     * @return number of element of const_buffer_sequence     */    static constexpr std::size_t num_of_const_buffer_sequence() {        return 1;    }    /**     * @brief Create one continuours buffer.     *        All sequence of buffers are concatinated.     *        It is useful to store to file/database.     * @return continuous buffer     */    std::string continuous_buffer() const {        return std::string(message_.data(), size());    }protected:    boost::container::static_vector<char, 2 + PacketIdBytes> const& message() const {        return message_;    }private:    boost::container::static_vector<char, 2 + PacketIdBytes> message_;};} // namespace detail_v3_1_1template <std::size_t PacketIdBytes>struct basic_puback_message : detail_v3_1_1::basic_header_packet_id_message<PacketIdBytes> {    using base = detail_v3_1_1::basic_header_packet_id_message<PacketIdBytes>;    basic_puback_message(typename packet_id_type<PacketIdBytes>::type packet_id)        : base(control_packet_type::puback, 0b0000, packet_id)    {}};using puback_message = basic_puback_message<2>;template <std::size_t PacketIdBytes>struct basic_pubrec_message : detail_v3_1_1::basic_header_packet_id_message<PacketIdBytes> {    using base = detail_v3_1_1::basic_header_packet_id_message<PacketIdBytes>;    basic_pubrec_message(typename packet_id_type<PacketIdBytes>::type packet_id)        : base(control_packet_type::pubrec, 0b0000, packet_id)    {}};using pubrec_message = basic_pubrec_message<2>;template <std::size_t PacketIdBytes>struct basic_pubrel_message : detail_v3_1_1::basic_header_packet_id_message<PacketIdBytes> {    using base = detail_v3_1_1::basic_header_packet_id_message<PacketIdBytes>;    basic_pubrel_message(typename packet_id_type<PacketIdBytes>::type packet_id)        : base(control_packet_type::pubrel, 0b0010, packet_id)    {    }    basic_pubrel_message(string_view buf)        : base(buf.begin(), buf.end())    {    }    /**     * @brief Get packet id     * @return packet_id     */    typename packet_id_type<PacketIdBytes>::type packet_id() const {        return make_packet_id<PacketIdBytes>::apply(std::next(base::message().begin(), 2), base::message().end());    }};using pubrel_message = basic_pubrel_message<2>;using pubrel_32_message = basic_pubrel_message<4>;template <std::size_t PacketIdBytes>struct basic_pubcomp_message : detail_v3_1_1::basic_header_packet_id_message<PacketIdBytes> {    basic_pubcomp_message(typename packet_id_type<PacketIdBytes>::type packet_id)        : detail_v3_1_1::basic_header_packet_id_message<PacketIdBytes>(control_packet_type::pubcomp, 0b0000, packet_id)    {}};using pubcomp_message = basic_pubcomp_message<2>;template <std::size_t PacketIdBytes>struct basic_unsuback_message : detail_v3_1_1::basic_header_packet_id_message<PacketIdBytes> {    basic_unsuback_message(typename packet_id_type<PacketIdBytes>::type packet_id)        : detail_v3_1_1::basic_header_packet_id_message<PacketIdBytes>(control_packet_type::unsuback, 0b0000, packet_id)    {}};using unsuback_message = basic_unsuback_message<2>;struct pingreq_message : detail_v3_1_1::header_only_message {    pingreq_message()        : detail_v3_1_1::header_only_message(control_packet_type::pingreq, 0b0000)    {}};struct pingresp_message : detail_v3_1_1::header_only_message {    pingresp_message()        : detail_v3_1_1::header_only_message(control_packet_type::pingresp, 0b0000)    {}};struct disconnect_message : detail_v3_1_1::header_only_message {    disconnect_message()        : detail_v3_1_1::header_only_message(control_packet_type::disconnect, 0b0000)    {}};class connack_message {public:    connack_message(bool session_present, connect_return_code return_code)        : message_ {              static_cast<char>(make_fixed_header(control_packet_type::connack, 0b0000)),              0b0010,              static_cast<char>(session_present ? 1 : 0),              static_cast<char>(return_code)          }    {}    /**     * @brief Create const buffer sequence     *        it is for boost asio APIs     * @return const buffer sequence     */    std::vector<as::const_buffer> const_buffer_sequence() const {        return { as::buffer(message_.data(), size()) };    }    /**     * @brief Get whole size of sequence     * @return whole size     */    std::size_t size() const {        return message_.size();    }    /**     * @brief Get number of element of const_buffer_sequence     * @return number of element of const_buffer_sequence     */    static constexpr std::size_t num_of_const_buffer_sequence() {        return 1;    }    /**     * @brief Create one continuours buffer.     *        All sequence of buffers are concatinated.     *        It is useful to store to file/database.     * @return continuous buffer     */    std::string continuous_buffer() const {        return std::string(message_.data(), size());    }private:    boost::container::static_vector<char, 4> message_;};// variable length messagesclass connect_message {public:    connect_message(        std::uint16_t keep_alive_sec,        buffer client_id,        bool clean_session,        optional<will> w,        optional<buffer> user_name,        optional<buffer> password    )        : fixed_header_(static_cast<char>(make_fixed_header(control_packet_type::connect, 0b0000))),          connect_flags_(0),          // protocol name length, protocol name, protocol level, connect flag, client id length, client id, keep alive          remaining_length_(              2 +                     // protocol name length              4 +                     // protocol name              1 +                     // protocol level              1 +                     // connect flag              2 +                     // keep alive              2 +                     // client id length              client_id.size()        // client id          ),          protocol_name_and_level_ { 0x00, 0x04, 'M', 'Q', 'T', 'T', 0x04 },          client_id_(force_move(client_id)),          client_id_length_buf_{ num_to_2bytes(boost::numeric_cast<std::uint16_t>(client_id_.size())) },          keep_alive_buf_ { num_to_2bytes(keep_alive_sec) }    {        utf8string_check(client_id_);        if (clean_session) connect_flags_ |= connect_flags::clean_session;        if (user_name) {            utf8string_check(user_name.value());            connect_flags_ |= connect_flags::user_name_flag;            user_name_ = force_move(user_name.value());            add_uint16_t_to_buf(user_name_length_buf_, boost::numeric_cast<std::uint16_t>(user_name_.size()));            remaining_length_ += 2 + user_name_.size();        }        if (password) {            connect_flags_ |= connect_flags::password_flag;            password_ = force_move(password.value());            add_uint16_t_to_buf(password_length_buf_, boost::numeric_cast<std::uint16_t>(password_.size()));            remaining_length_ += 2 + password_.size();        }        if (w) {            connect_flags_ |= connect_flags::will_flag;            if (w.value().get_retain() == retain::yes) connect_flags_ |= connect_flags::will_retain;            connect_flags::set_will_qos(connect_flags_, w.value().get_qos());            utf8string_check(w.value().topic());            will_topic_name_ = force_move(w.value().topic());            add_uint16_t_to_buf(                will_topic_name_length_buf_,                boost::numeric_cast<std::uint16_t>(will_topic_name_.size())            );            if (w.value().message().size() > 0xffffL) throw will_message_length_error();            will_message_ = force_move(w.value().message());            add_uint16_t_to_buf(                will_message_length_buf_,                boost::numeric_cast<std::uint16_t>(will_message_.size()));            remaining_length_ += 2 + will_topic_name_.size() + 2 + will_message_.size();        }        auto rb = remaining_bytes(remaining_length_);        for (auto e : rb) {            remaining_length_buf_.push_back(e);        }    }    /**     * @brief Create const buffer sequence     *        it is for boost asio APIs     * @return const buffer sequence     */    std::vector<as::const_buffer> const_buffer_sequence() const {        std::vector<as::const_buffer> ret;        ret.reserve(num_of_const_buffer_sequence());        ret.emplace_back(as::buffer(&fixed_header_, 1));        ret.emplace_back(as::buffer(remaining_length_buf_.data(), remaining_length_buf_.size()));        ret.emplace_back(as::buffer(protocol_name_and_level_.data(), protocol_name_and_level_.size()));        ret.emplace_back(as::buffer(&connect_flags_, 1));        ret.emplace_back(as::buffer(keep_alive_buf_.data(), keep_alive_buf_.size()));        ret.emplace_back(as::buffer(client_id_length_buf_.data(), client_id_length_buf_.size()));        ret.emplace_back(as::buffer(client_id_));        if (connect_flags::has_will_flag(connect_flags_)) {            ret.emplace_back(as::buffer(will_topic_name_length_buf_.data(), will_topic_name_length_buf_.size()));            ret.emplace_back(as::buffer(will_topic_name_));            ret.emplace_back(as::buffer(will_message_length_buf_.data(), will_message_length_buf_.size()));            ret.emplace_back(as::buffer(will_message_));        }        if (connect_flags::has_user_name_flag(connect_flags_)) {            ret.emplace_back(as::buffer(user_name_length_buf_.data(), user_name_length_buf_.size()));            ret.emplace_back(as::buffer(user_name_));        }        if (connect_flags::has_password_flag(connect_flags_)) {            ret.emplace_back(as::buffer(password_length_buf_.data(), password_length_buf_.size()));            ret.emplace_back(as::buffer(password_));        }        return ret;    }    /**     * @brief Get whole size of sequence     * @return whole size     */    std::size_t size() const {        return            1 +                            // fixed header            remaining_length_buf_.size() +            remaining_length_;    }    /**     * @brief Get number of element of const_buffer_sequence     * @return number of element of const_buffer_sequence     */    static constexpr std::size_t num_of_const_buffer_sequence() {        return            1 +                   // fixed header            1 +                   // remaining length            1 +                   // protocol name and level            1 +                   // connect flags            1 +                   // keep alive            2 +                   // client id length, client id            2 +                   // will topic name length, will topic name            2 +                   // will message length, will message            2 +                   // user name length, user name            2;                    // password length, password    }    /**     * @brief Create one continuours buffer.     *        All sequence of buffers are concatinated.     *        It is useful to store to file/database.     * @return continuous buffer     */    std::string continuous_buffer() const {        std::string ret;        ret.reserve(size());        ret.push_back(static_cast<char>(fixed_header_));        ret.append(remaining_length_buf_.data(), remaining_length_buf_.size());        ret.append(protocol_name_and_level_.data(), protocol_name_and_level_.size());        ret.push_back(connect_flags_);        ret.append(keep_alive_buf_.data(), keep_alive_buf_.size());        ret.append(client_id_length_buf_.data(), client_id_length_buf_.size());        ret.append(client_id_.data(), client_id_.size());        if (connect_flags::has_will_flag(connect_flags_)) {            ret.append(will_topic_name_length_buf_.data(), will_topic_name_length_buf_.size());            ret.append(will_topic_name_.data(), will_topic_name_.size());            ret.append(will_message_length_buf_.data(), will_message_length_buf_.size());            ret.append(will_message_.data(), will_message_.size());        }        if (connect_flags::has_user_name_flag(connect_flags_)) {            ret.append(user_name_length_buf_.data(), user_name_length_buf_.size());            ret.append(user_name_.data(), user_name_.size());        }        if (connect_flags::has_password_flag(connect_flags_)) {            ret.append(password_length_buf_.data(), password_length_buf_.size());            ret.append(password_.data(), password_.size());        }        return ret;    }private:    std::uint8_t fixed_header_;    char connect_flags_;    std::size_t remaining_length_;    boost::container::static_vector<char, 4> remaining_length_buf_;    boost::container::static_vector<char, 7> protocol_name_and_level_;    buffer client_id_;    boost::container::static_vector<char, 2> client_id_length_buf_;    buffer will_topic_name_;    boost::container::static_vector<char, 2> will_topic_name_length_buf_;    buffer will_message_;    boost::container::static_vector<char, 2> will_message_length_buf_;    buffer user_name_;    boost::container::static_vector<char, 2> user_name_length_buf_;    buffer password_;    boost::container::static_vector<char, 2> password_length_buf_;    boost::container::static_vector<char, 2> keep_alive_buf_;};template <std::size_t PacketIdBytes>class basic_publish_message {public:    template <        typename ConstBufferSequence,        typename std::enable_if<            as::is_const_buffer_sequence<ConstBufferSequence>::value,            std::nullptr_t        >::type = nullptr    >    basic_publish_message(        typename packet_id_type<PacketIdBytes>::type packet_id,        as::const_buffer topic_name,        ConstBufferSequence payloads,        publish_options pubopts    )        : fixed_header_(make_fixed_header(control_packet_type::publish, 0b0000) | pubopts.operator std::uint8_t()),          topic_name_(topic_name),          topic_name_length_buf_ { num_to_2bytes(boost::numeric_cast<std::uint16_t>(topic_name.size())) },          remaining_length_(              2                      // topic name length              + topic_name_.size()   // topic name              + (  (pubopts.get_qos() == qos::at_least_once || pubopts.get_qos() == qos::exactly_once)                 ? PacketIdBytes // packet_id                 : 0)          )    {        auto b = as::buffer_sequence_begin(payloads);        auto e = as::buffer_sequence_end(payloads);        auto num_of_payloads = static_cast<std::size_t>(std::distance(b, e));        payloads_.reserve(num_of_payloads);        for (; b != e; ++b) {            auto const& payload = *b;            remaining_length_ += payload.size();            payloads_.push_back(payload);        }        utf8string_check(topic_name_);        auto rb = remaining_bytes(remaining_length_);        for (auto e : rb) {            remaining_length_buf_.push_back(e);        }        if (pubopts.get_qos() == qos::at_least_once ||            pubopts.get_qos() == qos::exactly_once) {            packet_id_.reserve(PacketIdBytes);            add_packet_id_to_buf<PacketIdBytes>::apply(packet_id_, packet_id);        }    }    // Used in test code, and to deserialize stored messages.    basic_publish_message(buffer buf) {        if (buf.empty())  throw remaining_length_error();        fixed_header_ = static_cast<std::uint8_t>(buf.front());        qos qos_value = get_qos();        buf.remove_prefix(1);        if (buf.empty()) throw remaining_length_error();        auto len_consumed = remaining_length(buf.begin(), buf.end());        remaining_length_ = std::get<0>(len_consumed);        auto consumed = std::get<1>(len_consumed);        std::copy(            buf.begin(),            std::next(buf.begin(), static_cast<string_view::difference_type>(consumed)),            std::back_inserter(remaining_length_buf_));        buf.remove_prefix(consumed);        if (buf.size() < 2) throw remaining_length_error();        std::copy(buf.begin(), std::next(buf.begin(), 2), std::back_inserter(topic_name_length_buf_));        auto topic_name_length = make_uint16_t(topic_name_length_buf_.begin(), topic_name_length_buf_.end());        buf.remove_prefix(2);        if (buf.size() < topic_name_length) throw remaining_length_error();        topic_name_ = as::buffer(buf.substr(0, topic_name_length));        utf8string_check(topic_name_);        buf.remove_prefix(topic_name_length);        switch (qos_value) {        case qos::at_most_once:            break;        case qos::at_least_once:        case qos::exactly_once:            if (buf.size() < PacketIdBytes) throw remaining_length_error();            std::copy(buf.begin(), std::next(buf.begin(), PacketIdBytes), std::back_inserter(packet_id_));            buf.remove_prefix(PacketIdBytes);            break;        default:            throw protocol_error();            break;        };        if (!buf.empty()) {            payloads_.emplace_back(as::buffer(buf));        }    }    /**     * @brief Create const buffer sequence     *        it is for boost asio APIs     * @return const buffer sequence     */    std::vector<as::const_buffer> const_buffer_sequence() const {        std::vector<as::const_buffer> ret;        ret.reserve(num_of_const_buffer_sequence());        ret.emplace_back(as::buffer(&fixed_header_, 1));        ret.emplace_back(as::buffer(remaining_length_buf_.data(), remaining_length_buf_.size()));        ret.emplace_back(as::buffer(topic_name_length_buf_.data(), topic_name_length_buf_.size()));        ret.emplace_back(as::buffer(topic_name_));        if (!packet_id_.empty()) {            ret.emplace_back(as::buffer(packet_id_.data(), packet_id_.size()));        }        std::copy(payloads_.begin(), payloads_.end(), std::back_inserter(ret));        return ret;    }    /**     * @brief Get whole size of sequence     * @return whole size     */    std::size_t size() const {        return            1 +                            // fixed header            remaining_length_buf_.size() +            remaining_length_;    }    /**     * @brief Get number of element of const_buffer_sequence     * @return number of element of const_buffer_sequence     */    std::size_t num_of_const_buffer_sequence() const {        return            1 +                   // fixed header            1 +                   // remaining length            2 +                   // topic name length, topic name            (packet_id_.empty() ? 0 : 1) +  // packet_id            payloads_.size();    }    /**     * @brief Create one continuous buffer.     *        All sequence of buffers are concatinated.     *        It is useful to store to file/database.     * @return continuous buffer     */    std::string continuous_buffer() const {        std::string ret;        ret.reserve(size());        ret.push_back(static_cast<char>(fixed_header_));        ret.append(remaining_length_buf_.data(), remaining_length_buf_.size());        ret.append(topic_name_length_buf_.data(), topic_name_length_buf_.size());        ret.append(get_pointer(topic_name_), get_size(topic_name_));        ret.append(packet_id_.data(), packet_id_.size());        for (auto const& payload : payloads_) {            ret.append(get_pointer(payload), get_size(payload));        }        return ret;    }    /**     * @brief Get packet id     * @return packet_id     */    typename packet_id_type<PacketIdBytes>::type packet_id() const {        return make_packet_id<PacketIdBytes>::apply(packet_id_.begin(), packet_id_.end());    }    /**     * @brief Get publish_options     * @return publish_options.     */    constexpr publish_options get_options() const {        return publish_options(fixed_header_);    }    /**     * @brief Get qos     * @return qos     */    constexpr qos get_qos() const {        return publish::get_qos(fixed_header_);    }    /**     * @brief Check retain flag     * @return true if retain, otherwise return false.     */    constexpr bool is_retain() const {        return publish::is_retain(fixed_header_);    }    /**     * @brief Check dup flag     * @return true if dup, otherwise return false.     */    constexpr bool is_dup() const {        return publish::is_dup(fixed_header_);    }    /**     * @brief Get topic name     * @return topic name     */    constexpr string_view topic() const {        return string_view(get_pointer(topic_name_), get_size(topic_name_));    }    /**     * @brief Get payload     * @return payload     */    std::vector<string_view> payload() const {        std::vector<string_view> ret;        ret.reserve(payloads_.size());        for (auto const& payload : payloads_) {            ret.emplace_back(get_pointer(payload), get_size(payload));        }        return ret;    }    /**     * @brief Get payload as single buffer     * @return payload     */    buffer payload_as_buffer() const {        auto size = std::accumulate(            payloads_.begin(),            payloads_.end(),            std::size_t(0),            [](std::size_t s, as::const_buffer const& payload) {                return s += payload.size();            }        );        if (size == 0) return buffer();        auto spa = make_shared_ptr_array(size);        auto ptr = spa.get();        auto it = ptr;        for (auto const& payload : payloads_) {            auto b = get_pointer(payload);            auto s = get_size(payload);            auto e = b + s;            std::copy(b, e, it);            it += s;        }        return buffer(string_view(ptr, size), force_move(spa));    }    /**     * @brief Set dup flag     * @param dup flag value to set     */    constexpr void set_dup(bool dup) {        publish::set_dup(fixed_header_, dup);    }private:    std::uint8_t fixed_header_;    as::const_buffer topic_name_;    boost::container::static_vector<char, 2> topic_name_length_buf_;    boost::container::static_vector<char, PacketIdBytes> packet_id_;    std::vector<as::const_buffer> payloads_;    std::size_t remaining_length_;    boost::container::static_vector<char, 4> remaining_length_buf_;};using publish_message = basic_publish_message<2>;using publish_32_message = basic_publish_message<4>;template <std::size_t PacketIdBytes>class basic_subscribe_message {private:    struct entry {        entry(as::const_buffer topic_name, subscribe_options qos_value)            : topic_name_(topic_name),              topic_name_length_buf_ { num_to_2bytes(boost::numeric_cast<std::uint16_t>(topic_name_.size())) },              qos_(qos_value.get_qos())        {}        as::const_buffer topic_name_;        boost::container::static_vector<char, 2> topic_name_length_buf_;        qos qos_;    };public:    basic_subscribe_message(        std::vector<std::tuple<as::const_buffer, subscribe_options>> params,        typename packet_id_type<PacketIdBytes>::type packet_id    )        : fixed_header_(make_fixed_header(control_packet_type::subscribe, 0b0010)),          remaining_length_(PacketIdBytes)    {        add_packet_id_to_buf<PacketIdBytes>::apply(packet_id_, packet_id);        // Check for errors before allocating.        for (auto&& e : params) {            as::const_buffer topic_name = std::get<0>(e);            utf8string_check(topic_name);        }        entries_.reserve(params.size());        for (auto&& e : params) {            as::const_buffer topic_name = std::get<0>(e);            size_t size = topic_name.size();            entries_.emplace_back(topic_name, std::get<1>(e));            remaining_length_ +=                2 +                     // topic name length                size +                  // topic name                1;                      // means QoS        }        auto rb = remaining_bytes(remaining_length_);        for (auto e : rb) {            remaining_length_buf_.push_back(e);        }    }    /**     * @brief Create const buffer sequence     *        it is for boost asio APIs     * @return const buffer sequence     */    std::vector<as::const_buffer> const_buffer_sequence() const {        std::vector<as::const_buffer> ret;        ret.reserve(num_of_const_buffer_sequence());        ret.emplace_back(as::buffer(&fixed_header_, 1));        ret.emplace_back(as::buffer(remaining_length_buf_.data(), remaining_length_buf_.size()));        ret.emplace_back(as::buffer(packet_id_.data(), packet_id_.size()));        for (auto const& e : entries_) {            ret.emplace_back(as::buffer(e.topic_name_length_buf_.data(), e.topic_name_length_buf_.size()));            ret.emplace_back(as::buffer(e.topic_name_));            ret.emplace_back(as::buffer(&e.qos_, 1));        }        return ret;    }    /**     * @brief Get whole size of sequence     * @return whole size     */    std::size_t size() const {        return            1 +                            // fixed header            remaining_length_buf_.size() +            remaining_length_;    }    /**     * @brief Get number of element of const_buffer_sequence     * @return number of element of const_buffer_sequence     */    std::size_t num_of_const_buffer_sequence() const {        return            1 +                   // fixed header            1 +                   // remaining length            1 +                   // packet id            entries_.size() * 3;  // topic name length, topic name, qos    }    /**     * @brief Create one continuours buffer.     *        All sequence of buffers are concatinated.     *        It is useful to store to file/database.     * @return continuous buffer     */    std::string continuous_buffer() const {        std::string ret;        ret.reserve(size());        ret.push_back(static_cast<char>(fixed_header_));        ret.append(remaining_length_buf_.data(), remaining_length_buf_.size());        ret.append(packet_id_.data(), packet_id_.size());        for (auto const& e : entries_) {            ret.append(e.topic_name_length_buf_.data(), e.topic_name_length_buf_.size());            ret.append(get_pointer(e.topic_name_), get_size(e.topic_name_));            ret.push_back(static_cast<char>(e.qos_));        }        return ret;    }private:    std::uint8_t fixed_header_;    std::vector<entry> entries_;    boost::container::static_vector<char, PacketIdBytes> packet_id_;    std::size_t remaining_length_;    boost::container::static_vector<char, 4> remaining_length_buf_;};using subscribe_message = basic_subscribe_message<2>;template <std::size_t PacketIdBytes>class basic_suback_message {public:    basic_suback_message(        std::vector<suback_return_code> params,        typename packet_id_type<PacketIdBytes>::type packet_id    )        : fixed_header_(make_fixed_header(control_packet_type::suback, 0b0000)),          remaining_length_(params.size() + PacketIdBytes)    {        add_packet_id_to_buf<PacketIdBytes>::apply(packet_id_, packet_id);        auto rb = remaining_bytes(remaining_length_);        for (auto e : rb) {            remaining_length_buf_.push_back(e);        }        // TODO: We should be able to simply static-cast params.data() into a char*.        entries_.reserve(params.size());        for (auto e : params) {            entries_.push_back(static_cast<char>(e));        }    }    /**     * @brief Create const buffer sequence     *        it is for boost asio APIs     * @return const buffer sequence     */    std::vector<as::const_buffer> const_buffer_sequence() const {        std::vector<as::const_buffer> ret;        ret.reserve(num_of_const_buffer_sequence());        ret.emplace_back(as::buffer(&fixed_header_, 1));        ret.emplace_back(as::buffer(remaining_length_buf_.data(), remaining_length_buf_.size()));        ret.emplace_back(as::buffer(packet_id_.data(), packet_id_.size()));        ret.emplace_back(as::buffer(entries_));        return ret;    }    /**     * @brief Get whole size of sequence     * @return whole size     */    std::size_t size() const {        return            1 +                            // fixed header            remaining_length_buf_.size() +            remaining_length_;    }    /**     * @brief Get number of element of const_buffer_sequence     * @return number of element of const_buffer_sequence     */    static constexpr std::size_t num_of_const_buffer_sequence() {        return 4; // fixed header, remaining length, packet_id, entries    }    /**     * @brief Create one continuours buffer.     *        All sequence of buffers are concatinated.     *        It is useful to store to file/database.     * @return continuous buffer     */    std::string continuous_buffer() const {        std::string ret;        ret.reserve(size());        ret.push_back(static_cast<char>(fixed_header_));        ret.append(remaining_length_buf_.data(), remaining_length_buf_.size());        ret.append(packet_id_.data(), packet_id_.size());        ret.append(entries_);        return ret;    }private:    std::uint8_t fixed_header_;    std::string entries_;    boost::container::static_vector<char, PacketIdBytes> packet_id_;    std::size_t remaining_length_;    boost::container::static_vector<char, 4> remaining_length_buf_;};using suback_message = basic_suback_message<2>;template <std::size_t PacketIdBytes>class basic_unsubscribe_message {private:    struct entry {        entry(as::const_buffer topic_name)            : topic_name_(force_move(topic_name)),              topic_name_length_buf_ { num_to_2bytes(boost::numeric_cast<std::uint16_t>(topic_name_.size())) }        {}        as::const_buffer topic_name_;        boost::container::static_vector<char, 2> topic_name_length_buf_;    };public:    basic_unsubscribe_message(        std::vector<as::const_buffer> params,        typename packet_id_type<PacketIdBytes>::type packet_id    )        : fixed_header_(make_fixed_header(control_packet_type::unsubscribe, 0b0010)),          remaining_length_(PacketIdBytes)    {        add_packet_id_to_buf<PacketIdBytes>::apply(packet_id_, packet_id);        // Check for errors before allocating.        for (auto&& e : params) {            utf8string_check(e);        }        entries_.reserve(params.size());        for (auto&& e : params) {            entries_.emplace_back(e);            remaining_length_ +=                2 +          // topic name length                e.size();    // topic name        }        auto rb = remaining_bytes(remaining_length_);        for (auto e : rb) {            remaining_length_buf_.push_back(e);        }    }    /**     * @brief Create const buffer sequence     *        it is for boost asio APIs     * @return const buffer sequence     */    std::vector<as::const_buffer> const_buffer_sequence() const {        std::vector<as::const_buffer> ret;        ret.reserve(num_of_const_buffer_sequence());        ret.emplace_back(as::buffer(&fixed_header_, 1));        ret.emplace_back(as::buffer(remaining_length_buf_.data(), remaining_length_buf_.size()));        ret.emplace_back(as::buffer(packet_id_.data(), packet_id_.size()));        for (auto const& e : entries_) {            ret.emplace_back(as::buffer(e.topic_name_length_buf_.data(), e.topic_name_length_buf_.size()));            ret.emplace_back(as::buffer(e.topic_name_));        }        return ret;    }    /**     * @brief Get whole size of sequence     * @return whole size     */    std::size_t size() const {        return            1 +                            // fixed header            remaining_length_buf_.size() +            remaining_length_;    }    /**     * @brief Get number of element of const_buffer_sequence     * @return number of element of const_buffer_sequence     */    std::size_t num_of_const_buffer_sequence() const {        return            1 +                   // fixed header            1 +                   // remaining length            1 +                   // packet id            entries_.size() * 2;  // topic name length, topic name    }    /**     * @brief Create one continuours buffer.     *        All sequence of buffers are concatinated.     *        It is useful to store to file/database.     * @return continuous buffer     */    std::string continuous_buffer() const {        std::string ret;        ret.reserve(size());        ret.push_back(static_cast<char>(fixed_header_));        ret.append(remaining_length_buf_.data(), remaining_length_buf_.size());        ret.append(packet_id_.data(), packet_id_.size());        for (auto const& e : entries_) {            ret.append(e.topic_name_length_buf_.data(), e.topic_name_length_buf_.size());            ret.append(get_pointer(e.topic_name_), get_size(e.topic_name_));        }        return ret;    }private:    std::uint8_t fixed_header_;    std::vector<entry> entries_;    boost::container::static_vector<char, PacketIdBytes> packet_id_;    std::size_t remaining_length_;    boost::container::static_vector<char, 4> remaining_length_buf_;};using unsubscribe_message = basic_unsubscribe_message<2>;} // inline namespace v3_1_1} // namespace MQTT_NS#endif // MQTT_MESSAGE_HPP
 |