EIC code displayed by LXR master/​include/​boost/​mqtt5/​impl/​replies.hpp	Source navigation Diff markup Identifier search General search
	Version: master test Revert   Change

File indexing completed on 2025-09-17 08:38:22

 //
 // Copyright (c) 2023-2025 Ivica Siladic, Bruno Iljazovic, Korina Simicevic
 //
 // Distributed under the Boost Software License, Version 1.0.
 // (See accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt)
 //
 
 #ifndef BOOST_MQTT5_REPLIES_HPP
 #define BOOST_MQTT5_REPLIES_HPP
 
 #include <boost/mqtt5/types.hpp>
 
 #include <boost/mqtt5/detail/control_packet.hpp>
 #include <boost/mqtt5/detail/internal_types.hpp>
 
 #include <boost/asio/any_completion_handler.hpp>
 #include <boost/asio/any_io_executor.hpp>
 #include <boost/asio/async_result.hpp>
 #include <boost/asio/consign.hpp>
 #include <boost/asio/dispatch.hpp>
 #include <boost/asio/error.hpp>
 #include <boost/asio/post.hpp>
 #include <boost/asio/prepend.hpp>
 
 #include <algorithm>
 #include <chrono>
 #include <cstdint>
 #include <memory>
 #include <string>
 #include <vector>
 
 namespace boost::mqtt5::detail {
 
 namespace asio = boost::asio;
 
 class replies {
 public:
     using executor_type = asio::any_io_executor;
 private:
     using Signature = void (error_code, byte_citer, byte_citer);
 
     static constexpr auto max_reply_time = std::chrono::seconds(20);
 
     class reply_handler {
         asio::any_completion_handler<Signature> _handler;
         control_code_e _code;
         uint16_t _packet_id;
         std::chrono::time_point<std::chrono::system_clock> _ts;
     public:
         template <typename H>
         reply_handler(control_code_e code, uint16_t pid, H&& handler) :
             _handler(std::forward<H>(handler)), _code(code), _packet_id(pid),
             _ts(std::chrono::system_clock::now())
         {}
 
         reply_handler(reply_handler&&) = default;
         reply_handler(const reply_handler&) = delete;
 
         reply_handler& operator=(reply_handler&&) = default;
         reply_handler& operator=(const reply_handler&) = delete;
 
         void complete(
             error_code ec,
             byte_citer first = byte_citer {}, byte_citer last = byte_citer {}
         ) {
             std::move(_handler)(ec, first, last);
         }
 
         void complete_post(const executor_type& ex, error_code ec) {
             asio::post(
                 ex,
                 asio::prepend(
                     std::move(_handler), ec, byte_citer {}, byte_citer {}
                 )
             );
         }
 
         uint16_t packet_id() const noexcept {
             return _packet_id;
         }
 
         control_code_e code() const noexcept {
             return _code;
         }
 
         auto time() const noexcept {
             return _ts;
         }
     };
 
     executor_type _ex;
 
     using handlers = std::vector<reply_handler>;
     handlers _handlers;
 
     struct fast_reply {
         control_code_e code;
         uint16_t packet_id;
         std::unique_ptr<std::string> packet;
     };
     using fast_replies = std::vector<fast_reply>;
     fast_replies _fast_replies;
 
 public:
     template <typename Executor>
     explicit replies(Executor ex) : _ex(std::move(ex)) {}
 
     replies(replies&&) = default;
     replies(const replies&) = delete;
 
     replies& operator=(replies&&) = default;
     replies& operator=(const replies&) = delete;
 
     template <typename CompletionToken>
     decltype(auto) async_wait_reply(
         control_code_e code, uint16_t packet_id, CompletionToken&& token
     ) {
         auto dup_handler_ptr = find_handler(code, packet_id);
         if (dup_handler_ptr != _handlers.end()) {
             dup_handler_ptr->complete_post(_ex, asio::error::operation_aborted);
             _handlers.erase(dup_handler_ptr);
         }
 
         auto freply = find_fast_reply(code, packet_id);
 
         if (freply == _fast_replies.end()) {
             auto initiation = [](
                 auto handler, replies& self,
                 control_code_e code, uint16_t packet_id
             ) {
                 self._handlers.emplace_back(
                     code, packet_id, std::move(handler)
                 );
             };
             return asio::async_initiate<CompletionToken, Signature>(
                 initiation, token, std::ref(*this), code, packet_id
             );
         }
 
         auto fdata = std::move(*freply);
         _fast_replies.erase(freply);
 
         auto initiation = [](
             auto handler, std::unique_ptr<std::string> packet,
             const executor_type& ex
         ) {
             byte_citer first = packet->cbegin();
             byte_citer last = packet->cend();
 
             asio::post(
                 ex,
                 asio::consign(
                     asio::prepend(
                         std::move(handler), error_code {}, first, last
                     ),
                     std::move(packet)
                 )
             );
         };
 
         return asio::async_initiate<CompletionToken, Signature>(
             initiation, token, std::move(fdata.packet), _ex
         );
     }
 
     void dispatch(
         error_code ec, control_code_e code, uint16_t packet_id,
         byte_citer first, byte_citer last
     ) {
         auto handler_ptr = find_handler(code, packet_id);
 
         if (handler_ptr == _handlers.end()) {
             _fast_replies.push_back({
                 code, packet_id,
                 std::make_unique<std::string>(first, last)
             });
             return;
         }
 
         auto handler = std::move(*handler_ptr);
         _handlers.erase(handler_ptr);
         handler.complete(ec, first, last);
     }
 
     void resend_unanswered() {
         auto ua = std::move(_handlers);
         for (auto& h : ua)
             h.complete(asio::error::try_again);
     }
 
     void cancel_unanswered() {
         auto ua = std::move(_handlers);
         for (auto& h : ua)
             h.complete_post(_ex, asio::error::operation_aborted);
     }
 
     bool any_expired() {
         auto now = std::chrono::system_clock::now();
         return std::any_of(
             _handlers.begin(), _handlers.end(),
             [now](const auto& h) {
                 return now - h.time() > max_reply_time;
             }
         );
     }
 
     void clear_fast_replies() {
         _fast_replies.clear();
     }
 
     void clear_pending_pubrels() {
         for (auto it = _handlers.begin(); it != _handlers.end();) {
             if (it->code() == control_code_e::pubrel) {
                 it->complete(asio::error::operation_aborted);
                 it = _handlers.erase(it);
             }
             else
                 ++it;
         }
     }
 
 private:
     handlers::iterator find_handler(control_code_e code, uint16_t packet_id) {
         return std::find_if(
             _handlers.begin(), _handlers.end(),
             [code, packet_id](const auto& h) {
                 return h.code() == code && h.packet_id() == packet_id;
             }
         );
     }
 
     fast_replies::iterator find_fast_reply(
         control_code_e code, uint16_t packet_id
     ) {
         return std::find_if(
             _fast_replies.begin(), _fast_replies.end(),
             [code, packet_id](const auto& f) {
                 return f.code == code && f.packet_id == packet_id;
             }
         );
     }
 
 };
 
 } // end namespace boost::mqtt5::detail
 
 #endif // !BOOST_MQTT5_REPLIES_HPP

This page was automatically generated by the 2.3.7 LXR engine. The LXR team