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_1
- template <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 messages
- class 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
|