Wait for coroutines to stop on teardown()

include/libnetwrk/net/core/base_client.hpp

@@ -128,11 +128,16 @@ namespace libnetwrk {
 
     protected:
         void teardown() {
+            if (m_connection)
+                m_connection->stop();
+
+            /*
+                Wait for all coroutines to stop
+            */
+            wait_for_coroutines_to_stop();
+
             if (this->io_context && !this->io_context->stopped())
                 this->io_context->stop();
-
-            if (m_connection && m_connection->is_connected())
-                m_connection->stop();
 
             if (m_context_thread.joinable())
                 m_context_thread.join();
@@ -210,6 +215,15 @@ namespace libnetwrk {
             thread.detach();
         }
 
+        void wait_for_coroutines_to_stop() {
+            while (true) {
+                if (!m_connection)                        break;
+                if (m_connection->active_operations == 0) break;
+
+                std::this_thread::sleep_for(std::chrono::milliseconds(5));
+            }
+        }
+
         void ev_system_message(owned_message_t& msg) override final {
             system_command command = static_cast<system_command>(msg.msg.command());
 

include/libnetwrk/net/core/base_service.hpp

@@ -5,6 +5,7 @@
 
 #include <chrono>
 #include <list>
+#include <algorithm>
 
 namespace libnetwrk {
     template<typename Desc, typename Socket>
@@ -191,6 +192,16 @@ namespace libnetwrk {
             if (m_gc_future.valid())
                 m_gc_future.wait();
 
+            /*
+                Close all connections and signal coroutines to stop
+            */
+            stop_all_connections();
+
+            /*
+                Wait for all coroutines to stop
+            */
+            wait_for_coroutines_to_stop();
+
             if (this->io_context && !this->io_context->stopped())
                 this->io_context->stop();
 
@@ -301,6 +312,37 @@ namespace libnetwrk {
                 client->send(message);
         }
 
+        void stop_all_connections() {
+            std::lock_guard<std::mutex> guard(m_connections_mutex);
+
+            for (auto& client : m_connections) {
+                if (!client) continue;
+
+                client->stop();
+            }
+        }
+
+        void wait_for_coroutines_to_stop() {
+            bool running;
+
+            while (true) {
+                {
+                    std::lock_guard<std::mutex> guard(m_connections_mutex);
+
+                    running = std::any_of(m_connections.begin(), m_connections.end(),
+                        [](auto& client) {
+                            return client && client->active_operations != 0;
+                        }
+                    );
+                }
+
+                if (!running)
+                    break;
+
+                std::this_thread::sleep_for(std::chrono::milliseconds(5));
+            }
+        }
+
         asio::awaitable<void> co_gc() {
             auto current_executor = co_await asio::this_coro::executor;
             m_gc_timer            = std::make_unique<timer_t>(current_executor, std::chrono::seconds(gc_freq_sec));