Merge pull request #449 from redboltz/add_store_restore_to_client

Improved reconnect.

Author: redboltz

CHANGELOG.adoc

@@ -2,7 +2,10 @@
 
 = History
 
-== 10.2.9
+== 10.3.0
+* Added get_qos2_publish_handled_pids(), restore_qos2_publish_handled_pids(), restore_packets(), and get_stored_packets() to client for convenient. #449
+  * get_endpoint() can be omitted now.
+* Refined reconnect examples. #449
 * Added websocket async_close timeout. #448
 * Added documentation regarding stream reuse restrictions for TLS-related streams. #447
 

CMakeLists.txt

@@ -5,7 +5,7 @@
 # http://www.boost.org/LICENSE_1_0.txt)
 
 cmake_minimum_required (VERSION 3.13.0)
-project(async_mqtt_iface VERSION 10.2.9)
+project(async_mqtt_iface VERSION 10.3.0)
 
 set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)

README.md

@@ -3,7 +3,7 @@
 - I/O independent (also known as Sans-I/O) MQTT protocol library for C++17.
 - Asynchronous MQTT communication library using the MQTT protocol library and Boost.Asio.
 
-Version 10.2.9 [![Actions Status](https://github.com/redboltz/async_mqtt/workflows/CI/badge.svg)](https://github.com/redboltz/async_mqtt/actions)[![codecov](https://codecov.io/gh/redboltz/async_mqtt/branch/main/graph/badge.svg)](https://codecov.io/gh/redboltz/async_mqtt)
+Version 10.3.0 [![Actions Status](https://github.com/redboltz/async_mqtt/workflows/CI/badge.svg)](https://github.com/redboltz/async_mqtt/actions)[![codecov](https://codecov.io/gh/redboltz/async_mqtt/branch/main/graph/badge.svg)](https://codecov.io/gh/redboltz/async_mqtt)
 
 ## Document
 

example/cl_cpp17_mqtt_sub.cpp

@@ -32,7 +32,20 @@ struct app {
 
 private:
     void connect() {
-        cli_.async_underlying_handshake(
+        // For TLS (e.g. mqtts, wss), step 1, 2, and 3 are required due to Boost.Asio limitation.
+        // 1. Store MQTT session information.
+        auto packets = cli_->get_stored_packets();
+        auto pids = cli_->get_qos2_publish_handled_pids();
+
+        // 2. Re-construct client
+        cli_.emplace(cli_->get_executor());
+
+        // 3. Restore MQTT session information.
+        cli_->restore_packets(am::force_move(packets));
+        cli_->restore_qos2_publish_handled_pids(am::force_move(pids));
+
+        // 4. handshake
+        cli_->async_underlying_handshake(
             host_,
             port_,
             [this](auto&&... args) {
@@ -62,7 +75,7 @@ struct app {
             reconnect();
             return;
         }
-        cli_.async_start(
+        cli_->async_start(
             true,                // clean_start
             std::uint16_t(0),    // keep_alive
             "",                  // Client Identifier, empty means generated by the broker
@@ -95,8 +108,8 @@ struct app {
             {"topic2", am::qos::at_least_once},
             {"topic3", am::qos::exactly_once},
         };
-        cli_.async_subscribe(
-            *cli_.acquire_unique_packet_id(), // sync version only works thread safe context
+        cli_->async_subscribe(
+            *cli_->acquire_unique_packet_id(), // sync version only works thread safe context
             am::force_move(sub_entry),
             [this](auto&&... args) {
                 handle_subscribe_response(
@@ -118,7 +131,7 @@ struct app {
         if (suback_opt) {
             std::cout << *suback_opt << std::endl;
         }
-        cli_.async_recv(
+        cli_->async_recv(
             [this](auto&&... args) {
                 handle_recv(
                     std::forward<std::remove_reference_t<decltype(args)>>(args)...
@@ -139,7 +152,7 @@ struct app {
         BOOST_ASSERT(pv_opt);
         std::cout << *pv_opt << std::endl;
         // next receive
-        cli_.async_recv(
+        cli_->async_recv(
             [this](auto&&... args) {
                 handle_recv(
                     std::forward<std::remove_reference_t<decltype(args)>>(args)...
@@ -148,10 +161,10 @@ struct app {
         );
     }
 
-    client_t cli_;
+    std::optional<client_t> cli_;
     std::string host_;
     std::string port_;
-    as::steady_timer tim_{cli_.get_executor()};
+    as::steady_timer tim_{cli_->get_executor()};
 };
 
 int main(int argc, char* argv[]) {

example/cl_cpp20coro_mqtt_sub.cpp

@@ -29,23 +29,34 @@ using awaitable_client =
 
 as::awaitable<void>
 proc(
-    awaitable_client& amcl,
     std::string_view host,
     std::string_view port) {
 
     auto exe = co_await as::this_coro::executor;
 
     std::cout << "start" << std::endl;
 
+    std::optional<awaitable_client> amcl{exe};
     while (true) {
         try {
-            // all underlying layer handshaking
-            // (Resolve hostname, TCP handshake)
-            co_await amcl.async_underlying_handshake(host, port);
+            // For TLS (e.g. mqtts, wss), step 1, 2, and 3 are required due to Boost.Asio limitation.
+            // 1. Store MQTT session information.
+            auto packets = amcl->get_stored_packets();
+            auto pids = amcl->get_qos2_publish_handled_pids();
+
+            // 2. Re-construct client
+            amcl.emplace(amcl->get_executor());
+
+            // 3. Restore MQTT session information.
+            amcl->restore_packets(am::force_move(packets));
+            amcl->restore_qos2_publish_handled_pids(am::force_move(pids));
+
+            // 4. handshake
+            co_await amcl->async_underlying_handshake(host, port);
             std::cout << "mqtt undlerlying handshaked" << std::endl;
 
             // MQTT connect and receive loop start
-            auto connack_opt = co_await amcl.async_start(
+            auto connack_opt = co_await amcl->async_start(
                 true,                // clean_start
                 std::uint16_t(0),    // keep_alive
                 "",                  // Client Identifier, empty means generated by the broker
@@ -64,8 +75,8 @@ proc(
                 {"topic2", am::qos::at_least_once},
                 {"topic3", am::qos::exactly_once},
             };
-            auto suback_opt = co_await amcl.async_subscribe(
-                *amcl.acquire_unique_packet_id(), // sync version only works thread safe context
+            auto suback_opt = co_await amcl->async_subscribe(
+                *amcl->acquire_unique_packet_id(), // sync version only works thread safe context
                 am::force_move(sub_entry)
             );
             if (suback_opt) {
@@ -74,7 +85,7 @@ proc(
 
             // recv (coroutine)
             while (true) {
-                auto pv_opt = co_await amcl.async_recv(as::use_awaitable);
+                auto pv_opt = co_await amcl->async_recv(as::use_awaitable);
                 BOOST_ASSERT(pv_opt);
                 pv_opt->visit(
                     am::overload{
@@ -115,7 +126,6 @@ int main(int argc, char* argv[]) {
         return -1;
     }
     as::io_context ioc;
-    auto amcl = awaitable_client{ioc.get_executor()};
-    as::co_spawn(amcl.get_executor(), proc(amcl, argv[1], argv[2]), as::detached);
+    as::co_spawn(ioc.get_executor(), proc(argv[1], argv[2]), as::detached);
     ioc.run();
 }

include/async_mqtt/asio_bind/client.hpp

@@ -8,6 +8,8 @@
 #define ASYNC_MQTT_ASIO_BIND_CLIENT_HPP
 
 #include <optional>
+#include <set>
+#include <vector>
 #include <boost/asio/async_result.hpp>
 #include <boost/asio/any_io_executor.hpp>
 
@@ -17,6 +19,7 @@
 #include <async_mqtt/protocol/error.hpp>
 #include <async_mqtt/protocol/role.hpp>
 #include <async_mqtt/protocol/packet/packet_id_type.hpp>
+#include <async_mqtt/protocol/packet/store_packet_variant_fwd.hpp>
 #include <async_mqtt/asio_bind/endpoint_fwd.hpp>
 
 namespace async_mqtt {
@@ -835,6 +838,39 @@ class client {
      */
     void release_packet_id(packet_id_type packet_id);
 
+    /**
+     * @brief Get processed but not released QoS2 packet ids
+     *        This function should be called after disconnection
+     * @return set of packet_ids
+     */
+    std::set<packet_id_type> get_qos2_publish_handled_pids() const;
+
+    /**
+     * @brief Restore processed but not released QoS2 packet ids
+     *        This function should be called before receive the first publish
+     * @param pids packet ids
+     */
+    void restore_qos2_publish_handled_pids(std::set<packet_id_type> pids);
+
+    /**
+     * @brief restore packets
+     *        the restored packets would automatically send when CONNACK packet is received
+     * @param pvs packets to restore
+     */
+    void restore_packets(
+        std::vector<store_packet_variant> pvs
+    );
+
+    /**
+     * @brief get stored packets
+     *        sotred packets mean inflight packets.
+     *        @li PUBLISH packet (QoS1) not received PUBACK packet
+     *        @li PUBLISH packet (QoS1) not received PUBREC packet
+     *        @li PUBREL  packet not received PUBCOMP packet
+     * @return std::vector<store_packet_variant>
+     */
+    std::vector<store_packet_variant> get_stored_packets() const;
+
     /**
      * @brief rebinds the client type to another executor
      */

include/async_mqtt/asio_bind/impl/client_impl.hpp

@@ -81,6 +81,11 @@ class client_impl {
     bool register_packet_id(packet_id_type packet_id);
     void release_packet_id(packet_id_type packet_id);
 
+    std::set<packet_id_type> get_qos2_publish_handled_pids() const;
+    void restore_qos2_publish_handled_pids(std::set<packet_id_type> pids);
+    void restore_packets(std::vector<store_packet_variant> pvs);
+    std::vector<store_packet_variant> get_stored_packets() const;
+
     template <typename Executor1>
     struct rebind_executor {
         using other = client_impl<

include/async_mqtt/asio_bind/impl/client_misc.hpp

@@ -134,6 +134,34 @@ client_impl<Version, NextLayer>::set_read_buffer_size(std::size_t val) {
     ep_.set_read_buffer_size(val);
 }
 
+template <protocol_version Version, typename NextLayer>
+inline
+std::set<packet_id_type>
+client_impl<Version, NextLayer>::get_qos2_publish_handled_pids() const {
+    return ep_.get_qos2_publish_handled_pids();
+}
+
+template <protocol_version Version, typename NextLayer>
+inline
+void
+client_impl<Version, NextLayer>::restore_qos2_publish_handled_pids(std::set<packet_id_type> pids) {
+    ep_.restore_qos2_publish_handled_pids(force_move(pids));
+}
+
+template <protocol_version Version, typename NextLayer>
+inline
+void
+client_impl<Version, NextLayer>::restore_packets(std::vector<store_packet_variant> pvs) {
+    ep_.restore_packets(force_move(pvs));
+}
+
+template <protocol_version Version, typename NextLayer>
+inline
+std::vector<store_packet_variant>
+client_impl<Version, NextLayer>::get_stored_packets() const {
+    return ep_.get_stored_packets();
+}
+
 } // namespace detail
 
 // member functions
@@ -262,6 +290,38 @@ client<Version, NextLayer>::set_read_buffer_size(std::size_t val) {
     impl_->set_read_buffer_size(val);
 }
 
+template <protocol_version Version, typename NextLayer>
+inline
+std::set<packet_id_type>
+client<Version, NextLayer>::get_qos2_publish_handled_pids() const {
+    BOOST_ASSERT(impl_);
+    return impl_->get_qos2_publish_handled_pids();
+}
+
+template <protocol_version Version, typename NextLayer>
+inline
+void
+client<Version, NextLayer>::restore_qos2_publish_handled_pids(std::set<packet_id_type> pids) {
+    BOOST_ASSERT(impl_);
+    impl_->restore_qos2_publish_handled_pids(force_move(pids));
+}
+
+template <protocol_version Version, typename NextLayer>
+inline
+void
+client<Version, NextLayer>::restore_packets(std::vector<store_packet_variant> pvs) {
+    BOOST_ASSERT(impl_);
+    impl_->restore_packets(force_move(pvs));
+}
+
+template <protocol_version Version, typename NextLayer>
+inline
+std::vector<store_packet_variant>
+client<Version, NextLayer>::get_stored_packets() const {
+    BOOST_ASSERT(impl_);
+    return impl_->get_stored_packets();
+}
+
 } // namespace async_mqtt
 
 #if !defined(ASYNC_MQTT_SEPARATE_COMPILATION)