加协程

module/Local_Server/External_Database/External_Database.cpp

@@ -595,12 +595,13 @@ External_Resources_Manager::query_callsign_external_databases_coro(
 ucoro::awaitable<std::vector<External_Resource_Status>>
 External_Resources_Manager::external_database_status_coro() const
 {
-    co_return co_await await_external_database_callback<
-        std::vector<External_Resource_Status>>(
+    co_return co_await await_external_database_callback<std::vector<External_Resource_Status>>
+    (
         [this](Status_List_Callback callback)
         {
             async_external_database_status(std::move(callback));
-        });
+        }
+    );
 }
 
 void External_Resources_Manager::server(Global* g)
@@ -616,8 +617,10 @@ void External_Resources_Manager::server(Global* g)
     });
     svr.PostCoro(api + "get_external_database_status", [this](HTTP_Param) -> drogon::Task<>
     {
+
         auto status_list =
             co_await Ecap_Coro::to_drogon(external_database_status_coro());
+
         res->setBody(
             warp(external_resource_status_list_to_json(status_list))
             .to_json_string());
@@ -646,8 +649,7 @@ void External_Resources_Manager::server(Global* g)
         CHECK_JSON_PARAM
         HTTP_REQUIRE_VALUE(name, params.try_get_string("name"))
         HTTP_REQUIRE_VALUE(source_file, params.try_get_string("source_file"))
-        const auto result =
-            co_await Ecap_Coro::to_drogon(import_external_database_coro(name, source_file));
+        const auto result = co_await Ecap_Coro::to_drogon(import_external_database_coro(name, source_file));
         g->save();
         res->setBody(warp(external_resource_status_to_json(result)).to_json_string());
         co_return;

module/Local_Server/server/Ucoro_Drogon_Glue.h

@@ -0,0 +1,129 @@
+#pragma once
+
+#include <drogon/drogon.h>
+#include <trantor/net/EventLoop.h>
+#include <ucoro/awaitable.hpp>
+
+#include <coroutine>
+#include <exception>
+#include <memory>
+#include <optional>
+#include <type_traits>
+#include <utility>
+#include <variant>
+
+namespace Ecap_Coro {
+
+template <typename T>
+class Ucoro_Drogon_Awaiter {
+public:
+    explicit Ucoro_Drogon_Awaiter(ucoro::awaitable<T>&& task)
+        : state_(std::make_shared<State>()), task_(std::move(task)) {}
+
+    bool await_ready() const noexcept { return false; }
+
+    void await_suspend(std::coroutine_handle<> continuation) {
+        state_->continuation = continuation;
+        state_->loop = trantor::EventLoop::getEventLoopOfCurrentThread();
+        if (state_->loop == nullptr) {
+            state_->loop = drogon::app().getLoop();
+        }
+
+        auto state = state_;
+        state_->running_task.emplace(
+            std::move(task_).detach_with_callback(
+                [state](ucoro::traits::exception_with_result_t<T> result) mutable {
+                    state->result = std::move(result);
+                    auto resume = [state]() { state->continuation.resume(); };
+                    if (state->loop != nullptr) {
+                        state->loop->queueInLoop(std::move(resume));
+                    } else {
+                        resume();
+                    }
+                }));
+        state_->running_task->start();
+    }
+
+    T await_resume() {
+        if (std::holds_alternative<std::exception_ptr>(state_->result)) {
+            std::rethrow_exception(std::get<std::exception_ptr>(state_->result));
+        }
+        return std::move(std::get<T>(state_->result));
+    }
+
+private:
+    struct State {
+        trantor::EventLoop* loop{};
+        std::coroutine_handle<> continuation{};
+        ucoro::traits::exception_with_result_t<T> result{};
+        std::optional<ucoro::awaitable<void>> running_task{};
+    };
+
+    std::shared_ptr<State> state_;
+    ucoro::awaitable<T> task_;
+};
+
+template <>
+class Ucoro_Drogon_Awaiter<void> {
+public:
+    explicit Ucoro_Drogon_Awaiter(ucoro::awaitable<void>&& task)
+        : state_(std::make_shared<State>()), task_(std::move(task)) {}
+
+    bool await_ready() const noexcept { return false; }
+
+    void await_suspend(std::coroutine_handle<> continuation) {
+        state_->continuation = continuation;
+        state_->loop = trantor::EventLoop::getEventLoopOfCurrentThread();
+        if (state_->loop == nullptr) {
+            state_->loop = drogon::app().getLoop();
+        }
+
+        auto state = state_;
+        state_->running_task.emplace(
+            std::move(task_).detach_with_callback(
+                [state](std::exception_ptr exception) mutable {
+                    state->exception = exception;
+                    auto resume = [state]() { state->continuation.resume(); };
+                    if (state->loop != nullptr) {
+                        state->loop->queueInLoop(std::move(resume));
+                    } else {
+                        resume();
+                    }
+                }));
+        state_->running_task->start();
+    }
+
+    void await_resume() {
+        if (state_->exception) {
+            std::rethrow_exception(state_->exception);
+        }
+    }
+
+private:
+    struct State {
+        trantor::EventLoop* loop{};
+        std::coroutine_handle<> continuation{};
+        std::exception_ptr exception{};
+        std::optional<ucoro::awaitable<void>> running_task{};
+    };
+
+    std::shared_ptr<State> state_;
+    ucoro::awaitable<void> task_;
+};
+
+template <typename T>
+auto to_drogon(ucoro::awaitable<T>&& task) {
+    return Ucoro_Drogon_Awaiter<T>{std::move(task)};
+}
+
+template <typename T>
+drogon::Task<T> to_drogon_task(ucoro::awaitable<T>&& task) {
+    co_return co_await to_drogon(std::move(task));
+}
+
+inline drogon::Task<> to_drogon_task(ucoro::awaitable<void>&& task) {
+    co_await to_drogon(std::move(task));
+    co_return;
+}
+
+}  // namespace Ecap_Coro