Added CompletionToken support for async_disconnect.

Author: redboltz

include/mqtt/client.hpp

@@ -26,6 +26,7 @@
 #include <mqtt/callable_overlay.hpp>
 #include <mqtt/strand.hpp>
 #include <mqtt/null_strand.hpp>
+#include <mqtt/is_invocable.hpp>
 
 namespace MQTT_NS {
 
@@ -1444,30 +1445,51 @@ class client : public endpoint<std::mutex, std::lock_guard, PacketIdBytes> {
      * When the endpoint disconnects using disconnect(), a will won't send.<BR>
      * See https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901205<BR>
      * @param timeout after timeout elapsed, force_disconnect() is automatically called.
+     * @param reason_code
+     *        DISCONNECT Reason Code<BR>
+     *        See https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901208<BR>
+     *        3.14.2.1 Disconnect Reason Code
+     * @param props
+     *        Properties<BR>
+     *        See https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901209<BR>
+     *        3.14.2.2 DISCONNECT Properties
      * @param func A callback function that is called when async operation will finish.
      */
-    void async_disconnect(
+    template <
+        typename CompletionToken = async_handler_t,
+        typename std::enable_if_t<
+            is_invocable<CompletionToken, error_code>::value
+        >* = nullptr
+    >
+    auto async_disconnect(
         std::chrono::steady_clock::duration timeout,
-        async_handler_t func = async_handler_t()) {
+        v5::disconnect_reason_code reason_code,
+        v5::properties props,
+        CompletionToken&& token = async_handler_t{}
+    ) {
+
         if (ping_duration_ != std::chrono::steady_clock::duration::zero()) tim_ping_.cancel();
-        if (base::connected()) {
-            std::weak_ptr<this_type> wp(std::static_pointer_cast<this_type>(this->shared_from_this()));
-            tim_close_.expires_after(force_move(timeout));
-            tim_close_.async_wait(
-                [wp = force_move(wp)](error_code ec) {
-                    if (auto sp = wp.lock()) {
-                        if (!ec) {
-                            sp->socket()->post(
-                                [sp] {
-                                    sp->force_disconnect();
-                                }
-                            );
-                        }
+        std::weak_ptr<this_type> wp(std::static_pointer_cast<this_type>(this->shared_from_this()));
+        tim_close_.expires_after(force_move(timeout));
+        tim_close_.async_wait(
+            [wp = force_move(wp)](error_code ec) {
+                if (auto sp = wp.lock()) {
+                    if (!ec) {
+                        sp->socket()->post(
+                            [sp] {
+                                sp->force_disconnect();
+                            }
+                        );
                     }
                 }
+            }
+        );
+        return
+            base::async_disconnect(
+                reason_code,
+                force_move(props),
+                std::forward<CompletionToken>(token)
             );
-            base::async_disconnect(force_move(func));
-        }
     }
 
     /**
@@ -1476,7 +1498,6 @@ class client : public endpoint<std::mutex, std::lock_guard, PacketIdBytes> {
      * The broker disconnects the endpoint after receives the disconnect packet.<BR>
      * When the endpoint disconnects using disconnect(), a will won't send.<BR>
      * See https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901205<BR>
-     * @param timeout after timeout elapsed, force_disconnect() is automatically called.
      * @param reason_code
      *        DISCONNECT Reason Code<BR>
      *        See https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901208<BR>
@@ -1487,30 +1508,24 @@ class client : public endpoint<std::mutex, std::lock_guard, PacketIdBytes> {
      *        3.14.2.2 DISCONNECT Properties
      * @param func A callback function that is called when async operation will finish.
      */
-    void async_disconnect(
-        std::chrono::steady_clock::duration timeout,
+    template <
+        typename CompletionToken = async_handler_t,
+        typename std::enable_if_t<
+            is_invocable<CompletionToken, error_code>::value
+        >* = nullptr
+    >
+    auto async_disconnect(
         v5::disconnect_reason_code reason_code,
         v5::properties props,
-        async_handler_t func = async_handler_t()) {
+        CompletionToken&& token = async_handler_t{}
+    ) {
         if (ping_duration_ != std::chrono::steady_clock::duration::zero()) tim_ping_.cancel();
-        if (base::connected()) {
-            std::weak_ptr<this_type> wp(std::static_pointer_cast<this_type>(this->shared_from_this()));
-            tim_close_.expires_after(force_move(timeout));
-            tim_close_.async_wait(
-                [wp = force_move(wp)](error_code ec) {
-                    if (auto sp = wp.lock()) {
-                        if (!ec) {
-                            sp->socket()->post(
-                                [sp] {
-                                    sp->force_disconnect();
-                                }
-                            );
-                        }
-                    }
-                }
+        return
+            base::async_disconnect(
+                reason_code,
+                force_move(props),
+                std::forward<CompletionToken>(token)
             );
-            base::async_disconnect(reason_code, force_move(props), force_move(func));
-        }
     }
 
     /**
@@ -1519,22 +1534,39 @@ class client : public endpoint<std::mutex, std::lock_guard, PacketIdBytes> {
      * The broker disconnects the endpoint after receives the disconnect packet.<BR>
      * When the endpoint disconnects using disconnect(), a will won't send.<BR>
      * See https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901205<BR>
+     * @param timeout after timeout elapsed, force_disconnect() is automatically called.
      * @param func A callback function that is called when async operation will finish.
      */
-    template <typename CompletionToken>
+    template <
+        typename CompletionToken = async_handler_t,
+        typename std::enable_if_t<
+            is_invocable<CompletionToken, error_code>::value
+        >* = nullptr
+    >
     auto async_disconnect(
-        CompletionToken&& token = async_handler_t{}
-    )
-        ->
-        typename as::async_result<
-            typename std::decay<CompletionToken>::type,
-            void(error_code)
-        >::return_type
-    {
+        std::chrono::steady_clock::duration timeout,
+        CompletionToken&& token = async_handler_t{}) {
+
         if (ping_duration_ != std::chrono::steady_clock::duration::zero()) tim_ping_.cancel();
-        if (base::connected()) {
-            base::async_disconnect(std::forward<CompletionToken>(token));
-        }
+        std::weak_ptr<this_type> wp(std::static_pointer_cast<this_type>(this->shared_from_this()));
+        tim_close_.expires_after(force_move(timeout));
+        tim_close_.async_wait(
+            [wp = force_move(wp)](error_code ec) {
+                if (auto sp = wp.lock()) {
+                    if (!ec) {
+                        sp->socket()->post(
+                            [sp] {
+                                sp->force_disconnect();
+                            }
+                        );
+                    }
+                }
+            }
+        );
+        return
+            base::async_disconnect(
+                std::forward<CompletionToken>(token)
+            );
     }
 
     /**
@@ -1543,24 +1575,19 @@ class client : public endpoint<std::mutex, std::lock_guard, PacketIdBytes> {
      * The broker disconnects the endpoint after receives the disconnect packet.<BR>
      * When the endpoint disconnects using disconnect(), a will won't send.<BR>
      * See https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901205<BR>
-     * @param reason_code
-     *        DISCONNECT Reason Code<BR>
-     *        See https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901208<BR>
-     *        3.14.2.1 Disconnect Reason Code
-     * @param props
-     *        Properties<BR>
-     *        See https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901209<BR>
-     *        3.14.2.2 DISCONNECT Properties
      * @param func A callback function that is called when async operation will finish.
      */
-    void async_disconnect(
-        v5::disconnect_reason_code reason_code,
-        v5::properties props,
-        async_handler_t func = async_handler_t()) {
+    template <
+        typename CompletionToken = async_handler_t,
+        typename std::enable_if_t<
+            is_invocable<CompletionToken, error_code>::value
+        >* = nullptr
+    >
+    auto async_disconnect(
+        CompletionToken&& token = async_handler_t{}
+    ) {
         if (ping_duration_ != std::chrono::steady_clock::duration::zero()) tim_ping_.cancel();
-        if (base::connected()) {
-            base::async_disconnect(reason_code, force_move(props), force_move(func));
-        }
+        return base::async_disconnect(std::forward<CompletionToken>(token));
     }
 
     /**

include/mqtt/endpoint.hpp

@@ -72,6 +72,7 @@
 #include <mqtt/packet_id_manager.hpp>
 #include <mqtt/store.hpp>
 #include <mqtt/move_only_function.hpp>
+#include <mqtt/is_invocable.hpp>
 
 #if defined(MQTT_USE_WS)
 #include <mqtt/ws_endpoint.hpp>
@@ -2206,38 +2207,6 @@ class endpoint : public std::enable_shared_from_this<endpoint<Mutex, LockGuard,
         }
     }
 
-    struct async_disconnect_impl {
-        this_type& ep;
-        enum {
-            first,
-            second
-        } state = first;
-
-        template <typename Self>
-        void operator()(
-            Self& self
-        ) {
-            if (ep.connected_ && ep.mqtt_connected_) {
-                ep.disconnect_requested_ = true;
-                // The reason code and property vector are only used if we're using mqttv5.
-                ep.async_send_disconnect(
-                    v5::disconnect_reason_code::normal_disconnection,
-                    v5::properties{},
-                    [self = force_move(self)] (error_code ec) mutable {
-                        self.complete(ec);
-                    }
-                );
-            }
-            else {
-                ep.socket_->post(
-                    [self = force_move(self)] () mutable {
-                        self.complete(boost::system::errc::make_error_code(boost::system::errc::success));
-                    }
-                );
-            }
-        }
-    };
-
     /**
      * @brief Disconnect
      * @param func
@@ -2246,15 +2215,15 @@ class endpoint : public std::enable_shared_from_this<endpoint<Mutex, LockGuard,
      * The broker disconnects the endpoint after receives the disconnect packet.<BR>
      * When the endpoint disconnects using disconnect(), a will won't send.<BR>
      */
-    template <typename CompletionToken>
+    template <
+        typename CompletionToken = async_handler_t,
+        typename std::enable_if_t<
+            is_invocable<CompletionToken, error_code>::value
+        >* = nullptr
+    >
     auto async_disconnect(
         CompletionToken&& token = async_handler_t{}
-    )
-        ->
-        typename as::async_result<
-            typename std::decay<CompletionToken>::type,
-            void(error_code)
-        >::return_type {
+    ) {
         MQTT_LOG("mqtt_api", info)
             << MQTT_ADD_VALUE(address, this)
             << "async_disconnect";
@@ -2264,8 +2233,7 @@ class endpoint : public std::enable_shared_from_this<endpoint<Mutex, LockGuard,
                 void(error_code)
             >(
                 async_disconnect_impl{*this},
-                token,
-                get_executor()
+                token
             );
     }
 
@@ -2285,27 +2253,30 @@ class endpoint : public std::enable_shared_from_this<endpoint<Mutex, LockGuard,
      * The broker disconnects the endpoint after receives the disconnect packet.<BR>
      * When the endpoint disconnects using disconnect(), a will won't send.<BR>
      */
-    void async_disconnect(
+    template <
+        typename CompletionToken = async_handler_t,
+        typename std::enable_if_t<
+            is_invocable<CompletionToken, error_code>::value
+        >* = nullptr
+    >
+    auto async_disconnect(
         v5::disconnect_reason_code reason,
         v5::properties props = {},
-        async_handler_t func = {}
+        CompletionToken&& token = async_handler_t{}
     ) {
         MQTT_LOG("mqtt_api", info)
             << MQTT_ADD_VALUE(address, this)
             << "async_disconnect"
             << " reason:" << reason;
 
-        if (connected_ && mqtt_connected_) {
-            disconnect_requested_ = true;
-            async_send_disconnect(reason, force_move(props), force_move(func));
-        }
-        else {
-            socket_->post(
-                [func = force_move(func)] () mutable {
-                    if (func) func(boost::system::errc::make_error_code(boost::system::errc::success));
-                }
+        return
+            as::async_compose<
+                CompletionToken,
+                void(error_code)
+            >(
+                async_disconnect_impl{*this, reason, force_move(props)},
+                token
             );
-        }
     }
 
     /**
@@ -11218,6 +11189,36 @@ class endpoint : public std::enable_shared_from_this<endpoint<Mutex, LockGuard,
         }
     }
 
+    struct async_disconnect_impl {
+        this_type& ep;
+        v5::disconnect_reason_code reason = v5::disconnect_reason_code::normal_disconnection;
+        v5::properties props = {};
+
+        template <typename Self>
+        void operator()(
+            Self& self
+        ) {
+            if (ep.connected_ && ep.mqtt_connected_) {
+                ep.disconnect_requested_ = true;
+                // The reason code and property vector are only used if we're using mqttv5.
+                ep.async_send_disconnect(
+                    reason,
+                    force_move(props),
+                    [self = force_move(self)] (error_code ec) mutable {
+                        self.complete(ec);
+                    }
+                );
+            }
+            else {
+                ep.socket_->post(
+                    [self = force_move(self)] () mutable {
+                        self.complete(boost::system::errc::make_error_code(boost::system::errc::success));
+                    }
+                );
+            }
+        }
+    };
+
     void async_send_disconnect(
         v5::disconnect_reason_code reason,
         v5::properties props,