Separated internal queue.

include/async_mqtt/stream.hpp

@@ -27,6 +27,7 @@
 #include <async_mqtt/core/stream_traits.hpp>
 #include <async_mqtt/util/optional.hpp>
 #include <async_mqtt/util/static_vector.hpp>
+#include <async_mqtt/util/ioc_queue.hpp>
 #include <async_mqtt/buffer.hpp>
 #include <async_mqtt/constant.hpp>
 #include <async_mqtt/is_strand.hpp>
@@ -84,8 +85,6 @@ class stream : public std::enable_shared_from_this<stream<NextLayer>> {
                 )
             );
         }
-        queue_.emplace();
-        queue_->stop();
     }
 
     ~stream() {
@@ -372,7 +371,6 @@ class stream : public std::enable_shared_from_this<stream<NextLayer>> {
         std::shared_ptr<Packet> packet;
         error_code last_ec = error_code{};
         this_type_sp life_keeper = strm.shared_from_this();
-        optional<as::executor_work_guard<as::io_context::executor_type>> queue_work_guard = nullopt;
         enum { dispatch, post, write, bind, complete } state = dispatch;
 
         template <typename Self>
@@ -392,25 +390,18 @@ class stream : public std::enable_shared_from_this<stream<NextLayer>> {
                 BOOST_ASSERT(strm.strand_.running_in_this_thread());
                 state = write;
                 auto& a_strm{strm};
-                as::post(
-                    *a_strm.queue_,
+                a_strm.queue_.post(
                     as::bind_executor(
                         a_strm.strand_,
                         force_move(self)
                     )
                 );
-                if (!a_strm.writing_ && a_strm.queue_->stopped()) {
-                    a_strm.queue_->restart();
-                    a_strm.queue_->poll_one();
-                }
             } break;
             case write: {
                 BOOST_ASSERT(strm.strand_.running_in_this_thread());
+                strm.queue_.start_work();
                 if (strm.lowest_layer().is_open()) {
                     state = bind;
-                    BOOST_ASSERT(!strm.writing_);
-                    strm.writing_ = true;
-                    queue_work_guard.emplace(strm.queue_->get_executor());
                     auto& a_strm{strm};
                     auto cbs = packet->const_buffer_sequence();
                     async_write(
@@ -449,14 +440,13 @@ class stream : public std::enable_shared_from_this<stream<NextLayer>> {
         ) {
             if (ec) {
                 BOOST_ASSERT(strm.strand_.running_in_this_thread());
+                strm.queue_.stop_work();
                 auto& a_strm{strm};
                 as::post(
                     a_strm.strand_,
-                    [&a_strm, &queue = a_strm.queue_, wp = a_strm.weak_from_this()] {
+                    [&a_strm,wp = a_strm.weak_from_this()] {
                         if (auto sp = wp.lock()) {
-                            a_strm.writing_ = false;
-                            if (a_strm.queue_->stopped()) a_strm.queue_->restart();
-                            queue->poll_one();
+                            a_strm.queue_.poll_one();
                         }
                     }
                 );
@@ -476,14 +466,13 @@ class stream : public std::enable_shared_from_this<stream<NextLayer>> {
             switch (state) {
             case bind: {
                 BOOST_ASSERT(strm.strand_.running_in_this_thread());
+                strm.queue_.stop_work();
                 auto& a_strm{strm};
                 as::post(
                     a_strm.strand_,
-                    [&a_strm, &queue = a_strm.queue_, wp = a_strm.weak_from_this()] {
+                    [&a_strm, wp = a_strm.weak_from_this()] {
                         if (auto sp = wp.lock()) {
-                            a_strm.writing_ = false;
-                            if (a_strm.queue_->stopped()) a_strm.queue_->restart();
-                            queue->poll_one();
+                            a_strm.queue_.poll_one();
                         }
                     }
                 );
@@ -743,9 +732,8 @@ class stream : public std::enable_shared_from_this<stream<NextLayer>> {
 private:
     next_layer_type nl_;
     strand_type strand_{nl_.get_executor()};
-    optional<as::io_context> queue_;
+    ioc_queue queue_;
     static_vector<char, 5> header_remaining_length_buf_ = static_vector<char, 5>(5);
-    bool writing_ = false;
 };
 
 } // namespace async_mqtt

include/async_mqtt/util/ioc_queue.hpp

@@ -0,0 +1,63 @@
+// Copyright Takatoshi Kondo 2023
+//
+// 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(ASYNC_MQTT_UTIL_IOC_QUEUE_HPP)
+#define ASYNC_MQTT_UTIL_IOC_QUEUE_HPP
+
+#include <boost/asio.hpp>
+
+#include <async_mqtt/util/optional.hpp>
+
+namespace async_mqtt {
+
+namespace as = boost::asio;
+
+class ioc_queue {
+public:
+    ioc_queue() {
+        queue_.stop();
+    }
+
+    void start_work() {
+        working_ = true;
+        guard_.emplace(queue_.get_executor());
+    }
+
+    void stop_work() {
+        guard_.reset();
+    }
+
+    template <typename CompletionToken>
+    void post(CompletionToken&& token) {
+        as::post(
+            queue_,
+            std::forward<CompletionToken>(token)
+        );
+        if (!working_ && queue_.stopped()) {
+            queue_.restart();
+            queue_.poll_one();
+        }
+    }
+
+    bool stopped() const {
+        return queue_.stopped();
+    }
+
+    void poll_one() {
+        working_ = false;
+        if (queue_.stopped()) queue_.restart();
+        queue_.poll_one();
+    }
+
+private:
+    as::io_context queue_;
+    bool working_ = false;
+    optional<as::executor_work_guard<as::io_context::executor_type>> guard_;
+};
+
+} // namespace async_mqtt
+
+#endif // ASYNC_MQTT_UTIL_IOC_QUEUE_HPP