diff --git a/module/Local_Server/External_Database/External_Database.cpp b/module/Local_Server/External_Database/External_Database.cpp index f98f349..20b0918 100644 --- a/module/Local_Server/External_Database/External_Database.cpp +++ b/module/Local_Server/External_Database/External_Database.cpp @@ -595,12 +595,13 @@ External_Resources_Manager::query_callsign_external_databases_coro( ucoro::awaitable> External_Resources_Manager::external_database_status_coro() const { - co_return co_await await_external_database_callback< - std::vector>( + co_return co_await await_external_database_callback> + ( [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; diff --git a/module/Local_Server/server/Ucoro_Drogon_Glue.h b/module/Local_Server/server/Ucoro_Drogon_Glue.h new file mode 100644 index 0000000..fbab0ad --- /dev/null +++ b/module/Local_Server/server/Ucoro_Drogon_Glue.h @@ -0,0 +1,129 @@ +#pragma once + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace Ecap_Coro { + +template +class Ucoro_Drogon_Awaiter { +public: + explicit Ucoro_Drogon_Awaiter(ucoro::awaitable&& task) + : state_(std::make_shared()), 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 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(state_->result)) { + std::rethrow_exception(std::get(state_->result)); + } + return std::move(std::get(state_->result)); + } + +private: + struct State { + trantor::EventLoop* loop{}; + std::coroutine_handle<> continuation{}; + ucoro::traits::exception_with_result_t result{}; + std::optional> running_task{}; + }; + + std::shared_ptr state_; + ucoro::awaitable task_; +}; + +template <> +class Ucoro_Drogon_Awaiter { +public: + explicit Ucoro_Drogon_Awaiter(ucoro::awaitable&& task) + : state_(std::make_shared()), 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> running_task{}; + }; + + std::shared_ptr state_; + ucoro::awaitable task_; +}; + +template +auto to_drogon(ucoro::awaitable&& task) { + return Ucoro_Drogon_Awaiter{std::move(task)}; +} + +template +drogon::Task to_drogon_task(ucoro::awaitable&& task) { + co_return co_await to_drogon(std::move(task)); +} + +inline drogon::Task<> to_drogon_task(ucoro::awaitable&& task) { + co_await to_drogon(std::move(task)); + co_return; +} + +} // namespace Ecap_Coro