diff --git a/include/rest_rpc/asio_util.hpp b/include/rest_rpc/asio_util.hpp index 78765ba..393e906 100644 --- a/include/rest_rpc/asio_util.hpp +++ b/include/rest_rpc/asio_util.hpp @@ -1,11 +1,23 @@ #pragma once +#include "traits.h" #include "use_asio.hpp" namespace rest_rpc { +template +constexpr inline bool is_awaitable_v = + util::is_specialization_v, asio::awaitable>; + +template > struct return_type { + using type = T; +}; + +template struct return_type { + using type = typename std::remove_cvref_t::value_type; +}; + +template using return_type_t = typename return_type::type; + template inline auto async_start(auto executor, Coro &&coro) { - using R = - typename std::remove_cvref_t>::value_type; - static_assert(std::is_void_v); asio::co_spawn(executor, std::forward(coro), asio::detached); } diff --git a/include/rest_rpc/error_code.h b/include/rest_rpc/error_code.h index 51e83cd..62a4e1f 100644 --- a/include/rest_rpc/error_code.h +++ b/include/rest_rpc/error_code.h @@ -20,6 +20,7 @@ enum class rpc_errc : std::int8_t { protocol_error, has_response, duplicate_topic, + rpc_context_init_failed, }; class rpc_error_category : public std::error_category { @@ -60,6 +61,9 @@ public: return "has response, duplicate response is not allowed"; case rpc_errc::duplicate_topic: return "duplicate topic"; + case rpc_errc::rpc_context_init_failed: + return "the rpc context init failed, it must be created in rpc handler " + "io thread, otherwise will init failed"; default: return "unknown error"; } diff --git a/include/rest_rpc/rpc_client.hpp b/include/rest_rpc/rpc_client.hpp index e3c1a08..a53e82a 100644 --- a/include/rest_rpc/rpc_client.hpp +++ b/include/rest_rpc/rpc_client.hpp @@ -94,13 +94,15 @@ public: } template - asio::awaitable>> + asio::awaitable< + call_result>>> call(Args &&...args) { return call_for(std::chrono::seconds(5), std::forward(args)...); } template - asio::awaitable>> + asio::awaitable< + call_result>>> call_for(auto duration, Args &&...args) { using args_tuple = function_parameters_t; static_assert(std::is_constructible_v, @@ -108,7 +110,7 @@ public: rest_rpc_header header{}; header.function_id = get_key(); - using R = function_return_type_t; + using R = return_type_t>; auto r = co_await (watchdog(duration) || call_impl(header, std::forward(args)...)); if (r.index() == 0) { diff --git a/include/rest_rpc/rpc_connection.hpp b/include/rest_rpc/rpc_connection.hpp index 00f411c..4b34743 100644 --- a/include/rest_rpc/rpc_connection.hpp +++ b/include/rest_rpc/rpc_connection.hpp @@ -8,39 +8,42 @@ namespace rest_rpc { class rpc_connection; -class rpc_context { +class tls_data { public: - static auto &context() { - thread_local rpc_context instance; - return instance; - } - - rpc_context(rpc_context &&o) : conn_(std::move(o.conn_)), delay_(o.delay_) {} - rpc_context(const rpc_context &o) = delete; - void set_connection(std::shared_ptr conn) { conn_ = conn; } bool delay() { return delay_; } - void set_delay(bool r) { context().delay_ = r; } + void set_delay(bool r) { delay_ = r; } + + auto get_executor(); + std::shared_ptr get_conn() { return conn_; } + +private: + std::shared_ptr conn_ = nullptr; + bool delay_ = false; +}; + +inline auto &get_context() { + static thread_local tls_data instance; + return instance; +} + +class rpc_context { +public: + rpc_context(); auto get_executor(); template asio::awaitable response_s(Args &&...args); - template - std::error_code sync_response_s(Args &&...args); - template asio::awaitable response(Args &&...args); - template std::error_code sync_response(Args &&...args); - private: - rpc_context() = default; - std::shared_ptr conn_; - bool delay_ = false; + asio::any_io_executor executor_; + std::shared_ptr conn_ = nullptr; bool has_response_ = false; }; @@ -99,11 +102,11 @@ public: } // route - rpc_context::context().set_connection(self); + get_context().set_connection(self); auto result = co_await router_.route(header.function_id, body_); - bool delay = rpc_context::context().delay(); + bool delay = get_context().delay(); if (delay) { - rpc_context::context().set_delay(false); + get_context().set_delay(false); continue; } @@ -198,6 +201,21 @@ private: std::atomic topic_id_; }; +auto tls_data::get_executor() { + if (!conn_) { + return asio::any_io_executor(); + } + return conn_->get_executor(); +} + +auto rpc_context::get_executor() { return executor_; } + +rpc_context::rpc_context() { + executor_ = get_context().get_executor(); + conn_ = get_context().get_conn(); + get_context().set_delay(true); +} + // zero or one arguments template asio::awaitable rpc_context::response_s(Args &&...args) { @@ -210,29 +228,19 @@ asio::awaitable rpc_context::response_s(Args &&...args) { return response(std::forward(args)...); } -template -std::error_code rpc_context::sync_response_s(Args &&...args) { - return sync_wait(conn_->get_executor(), - response_s(std::forward(args)...)); -} - template asio::awaitable rpc_context::response(Args &&...args) { if (has_response_) { co_return make_error_code(rpc_errc::has_response); } - - has_response_ = true; + if (!conn_) { + REST_LOG_ERROR << "rpc context init failed"; + co_return make_error_code(rpc_errc::rpc_context_init_failed); + } rpc_service::msgpack_codec codec; rpc_result result(codec.pack_args(std::forward(args)...)); + has_response_ = true; co_return co_await conn_->response(result); } -template -std::error_code rpc_context::sync_response(Args &&...args) { - return sync_wait(conn_->get_executor(), - response(std::forward(args)...)); -} - -auto rpc_context::get_executor() { return conn_->get_executor(); } } // namespace rest_rpc \ No newline at end of file diff --git a/include/rest_rpc/rpc_router.hpp b/include/rest_rpc/rpc_router.hpp index 8b68a1e..6d19f0e 100644 --- a/include/rest_rpc/rpc_router.hpp +++ b/include/rest_rpc/rpc_router.hpp @@ -2,6 +2,7 @@ #include "codec.h" #include "error_code.h" +#include "asio_util.hpp" #include "util.hpp" #include #include @@ -9,10 +10,6 @@ #include namespace rest_rpc { -template -constexpr inline bool is_awaitable_v = - util::is_specialization_v, asio::awaitable>; - template constexpr inline bool is_void_v = std::is_same_v || std::is_same_v>; diff --git a/tests/test_rest_rpc.cpp b/tests/test_rest_rpc.cpp index 54b339d..ee7c0e1 100644 --- a/tests/test_rest_rpc.cpp +++ b/tests/test_rest_rpc.cpp @@ -69,9 +69,8 @@ template asio::awaitable response(auto ctx) { } std::string_view delay_response(std::string_view str) { - auto &ctx = rpc_context::context(); + rpc_context ctx; // set_delay before response in another thread - ctx.set_delay(true); // std::thread thd([ctx = std::move(ctx)]() mutable { // std::this_thread::sleep_for(std::chrono::seconds(2)); @@ -115,6 +114,66 @@ asio::awaitable no_arg_coro1() { co_return "test"; } +std::string delay_response1(std::string_view str) { + rpc_context ctx; // right, created in io thread. + async_start(ctx.get_executor(), ctx.response(std::string(str))); + + return ""; +} + +asio::awaitable delay_response2(std::string_view str) { + auto coro = []() -> asio::awaitable { + rpc_context ctx; // rpc_context init will be failed, it should be defined in + // io thread. + auto ret = co_await ctx.response("test"); + REST_LOG_INFO << ret.message(); + CHECK(ret); + }; + sync_wait(get_global_executor(), coro()); + + rpc_context ctx; + co_await ctx.response(str); + auto ret = co_await ctx.response(str); + CHECK(ret); + + co_return ""; +} + +std::string delay_response3(std::string str) { + rpc_context ctx; + std::thread thd([ctx = std::move(ctx), str]() mutable { + sync_wait(ctx.get_executor(), ctx.response(std::move(str))); + }); + thd.detach(); + + return ""; +} + +TEST_CASE("test delay response") { + using T = return_type_t; + rpc_server server("127.0.0.1:9005"); + server.register_handler(); + server.register_handler(); + server.register_handler(); + server.async_start(); + rpc_client client; + sync_wait(client.get_executor(), client.connect("127.0.0.1:9005")); + auto result = + sync_wait(client.get_executor(), client.call("test")); + CHECK(result.value == "test"); + auto result1 = + sync_wait(client.get_executor(), client.call("test")); + CHECK(result1.value == "test"); + result1 = + sync_wait(client.get_executor(), client.call("test")); + CHECK(result1.value == "test"); + auto result2 = sync_wait( + client.get_executor(), + client.call_for(std::chrono::minutes(2), "test")); + CHECK(result2.value == "test"); + server.stop(); +} + // TODO: client pool asio::awaitable test_router() { rpc_router router;