From 80b7e536ad9b5a6480d1e790023df47165081c92 Mon Sep 17 00:00:00 2001 From: qicosmos Date: Wed, 24 Sep 2025 19:22:09 +0800 Subject: [PATCH] reg coro --- include/rest_rpc/asio_util.hpp | 1 - include/rest_rpc/error_code.h | 5 +- include/rest_rpc/rpc_connection.hpp | 29 ++-- include/rest_rpc/rpc_router.hpp | 236 ++++++++++++++++------------ tests/test_rest_rpc1.cpp | 107 ++++++++++--- 5 files changed, 240 insertions(+), 138 deletions(-) diff --git a/include/rest_rpc/asio_util.hpp b/include/rest_rpc/asio_util.hpp index c983a39..0638abc 100644 --- a/include/rest_rpc/asio_util.hpp +++ b/include/rest_rpc/asio_util.hpp @@ -4,7 +4,6 @@ namespace rest_rpc { 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 042f764..997375e 100644 --- a/include/rest_rpc/error_code.h +++ b/include/rest_rpc/error_code.h @@ -17,7 +17,8 @@ enum class rpc_errc : std::int8_t { resolve_timeout, connection_timeout, request_timeout, - protocol_error + protocol_error, + has_response, }; class rpc_error_category : public std::error_category { @@ -54,6 +55,8 @@ public: return "request timeout"; case rpc_errc::protocol_error: return "protocol error"; + case rpc_errc::has_response: + return "has response, duplicate response is not allowed"; default: return "unknown error"; } diff --git a/include/rest_rpc/rpc_connection.hpp b/include/rest_rpc/rpc_connection.hpp index dff2164..ccf93ed 100644 --- a/include/rest_rpc/rpc_connection.hpp +++ b/include/rest_rpc/rpc_connection.hpp @@ -15,26 +15,28 @@ public: return instance; } - rpc_context(rpc_context&& o) : conn_(std::move(o.conn_)), delay_(o.delay_) {} - rpc_context(const rpc_context& o) = delete; + 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; } - + auto get_executor(); template asio::awaitable response(Args &&...args); - template std::error_code sync_response(Args &&...args); + template + std::error_code sync_response(Args &&...args); private: rpc_context() = default; std::shared_ptr conn_; bool delay_ = false; + bool has_response_ = false; }; class rpc_connection : public std::enable_shared_from_this { @@ -80,7 +82,7 @@ public: // route rpc_context::context().set_connection(self); - auto result = router_.route(header.function_id, body_); + auto result = co_await router_.route(header.function_id, body_); bool delay = rpc_context::context().delay(); if (delay) { rpc_context::context().set_delay(false); @@ -129,8 +131,14 @@ private: template asio::awaitable rpc_context::response(Args &&...args) { using args_tuple = typename function_traits::return_type; - static_assert(std::is_constructible_v, - "rpc function return type and response arguments are not match"); + static_assert( + std::is_constructible_v, + "rpc function return type and response arguments are not match"); + if (has_response_) { + co_return make_error_code(rpc_errc::has_response); + } + + has_response_ = true; rpc_service::msgpack_codec codec; rpc_result result(codec.pack_args(std::forward(args)...)); co_return co_await conn_->response(result); @@ -138,10 +146,9 @@ asio::awaitable rpc_context::response(Args &&...args) { template std::error_code rpc_context::sync_response(Args &&...args) { - return sync_wait(conn_->get_executor(), response(std::forward(args)...)); + return sync_wait(conn_->get_executor(), + response(std::forward(args)...)); } -auto rpc_context::get_executor() { - return conn_->get_executor(); -} +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 21fc3cb..f4d40dd 100644 --- a/include/rest_rpc/rpc_router.hpp +++ b/include/rest_rpc/rpc_router.hpp @@ -10,6 +10,14 @@ #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>; + struct rpc_result { rpc_result(std::string str) : result(std::move(str)) {} rpc_result(std::string_view str) : view(str) {} @@ -65,7 +73,7 @@ public: return std::to_string(key); } - rpc_result route(uint32_t key, std::string_view data) { + asio::awaitable route(uint32_t key, std::string_view data) { rpc_result route_result{}; try { rpc_service::msgpack_codec codec; @@ -74,7 +82,7 @@ public: route_result.result = "unknown function: " + get_name_by_key(key); route_result.ec = rpc_errc::no_such_function; } else { - it->second(data, route_result); + co_await it->second(data, route_result); route_result.ec = rpc_errc::ok; } } catch (const std::exception &ex) { @@ -89,71 +97,39 @@ public: route_result.ec = rpc_errc::function_unknown_exception; } - return route_result; + co_return route_result; } private: template - auto register_handler_impl(uint32_t key, std::string_view name, + void register_handler_impl(uint32_t key, std::string_view name, const Function &f, Self *self = nullptr) { if (key2func_name_.find(key) != key2func_name_.end()) { throw std::invalid_argument("duplicate registration key !"); - } else { - key2func_name_.emplace(key, name); - if constexpr (std::is_void_v) { - return register_nonmember_func(key, f); - } else { - return register_member_func(key, f, self); - } } - } - template - void register_nonmember_func(uint32_t key, Function f) { - this->map_invokers_[key] = [this, - f = std::move(f)](std::string_view str, - rpc_result &ret) mutable { - using args_tuple = typename function_traits::tuple_type; - using R = typename function_traits::return_type; + key2func_name_.emplace(key, name); - try { - if constexpr (std::tuple_size_v == 0) { - handle_zero_arg(f, ret); - } else { - using first_t = std::tuple_element_t<0, args_tuple>; - if constexpr (std::tuple_size_v == 1 && - util::is_basic_v) { - handle_one_arg(str, f, ret); - } else { - handle_more_args(str, f, ret); - } - } - } catch (std::invalid_argument &e) { - ret.ec = rpc_errc::invalid_argument; - ret.result = e.what(); - } catch (const std::exception &e) { - ret.ec = rpc_errc::function_exception; - ret.result = e.what(); - } - }; + register_func_impl(key, f, self); } template - void register_member_func(uint32_t key, const Function &f, Self *self) { - this->map_invokers_[key] = [this, f, self](std::string_view str, - rpc_result &ret) { + void register_func_impl(uint32_t key, const Function &f, Self *self) { + this->map_invokers_[key] = + [this, f, self](std::string_view str, + rpc_result &ret) -> asio::awaitable { using args_tuple = typename function_traits::tuple_type; - using R = typename function_traits::return_type; + using R = typename util::function_traits::return_type; try { if constexpr (std::tuple_size_v == 0) { - handle_zero_arg(f, self, ret); + co_await handle_zero_arg(f, ret, self); } else { using first_t = std::tuple_element_t<0, args_tuple>; if constexpr (std::tuple_size_v == 1 && util::is_basic_v) { - handle_one_arg(str, f, self, ret); + co_await handle_one_arg(str, f, ret, self); } else { - handle_more_args(str, f, self, ret); + co_await handle_more_args(str, f, ret, self); } } } catch (std::invalid_argument &e) { @@ -166,80 +142,134 @@ private: }; } - template - void handle_zero_arg(const F &f, rpc_result &ret) { - if constexpr (std::is_void_v) { - f(); - } else { - ret = rpc_service::msgpack_codec::pack_args(f()); - } - } - template - void handle_zero_arg(const F &f, Self *self, rpc_result &ret) { - if constexpr (std::is_void_v) { - (*self.*f)(); + asio::awaitable handle_zero_arg(const F &f, rpc_result &ret, + Self *self) { + if constexpr (is_void_v) { + if constexpr (std::is_void_v) { + if constexpr (is_awaitable_v) { + co_await f(); + } else { + f(); + } + } else { + if constexpr (is_awaitable_v) { + co_await (*self.*f)(); + } else { + (*self.*f)(); + } + } } else { - ret = rpc_service::msgpack_codec::pack_args((*self.*f)()); - } - } - - template - void handle_one_arg(std::string_view str, const F &f, rpc_result &ret) { - rpc_service::msgpack_codec codec; - if constexpr (std::is_void_v) { - f(codec.unpack(str)); - } else { - ret = rpc_service::msgpack_codec::pack_args(f(codec.unpack(str))); + if constexpr (std::is_void_v) { + if constexpr (is_awaitable_v) { + ret = rpc_service::msgpack_codec::pack_args(co_await f()); + } else { + ret = rpc_service::msgpack_codec::pack_args(f()); + } + } else { + if constexpr (is_awaitable_v) { + ret = rpc_service::msgpack_codec::pack_args(co_await (*self.*f)()); + } else { + ret = rpc_service::msgpack_codec::pack_args((*self.*f)()); + } + } } } template - void handle_one_arg(std::string_view str, const F &f, Self *self, - rpc_result &ret) { + asio::awaitable handle_one_arg(std::string_view str, const F &f, + rpc_result &ret, Self *self) { rpc_service::msgpack_codec codec; - if constexpr (std::is_void_v) { - (*self.*f)(codec.unpack(str)); + if constexpr (is_void_v) { + if constexpr (std::is_void_v) { + if constexpr (is_awaitable_v) { + co_await f(codec.unpack(str)); + } else { + f(codec.unpack(str)); + } + } else { + if constexpr (is_awaitable_v) { + co_await (*self.*f)(codec.unpack(str)); + } else { + (*self.*f)(codec.unpack(str)); + } + } } else { - ret = rpc_service::msgpack_codec::pack_args( - (*self.*f)(codec.unpack(str))); + if constexpr (std::is_void_v) { + if constexpr (is_awaitable_v) { + ret = rpc_service::msgpack_codec::pack_args( + co_await f(codec.unpack(str))); + } else { + ret = + rpc_service::msgpack_codec::pack_args(f(codec.unpack(str))); + } + } else { + if constexpr (is_awaitable_v) { + ret = rpc_service::msgpack_codec::pack_args( + co_await (*self.*f)(codec.unpack(str))); + } else { + ret = rpc_service::msgpack_codec::pack_args( + (*self.*f)(codec.unpack(str))); + } + } } } - template - void handle_more_args(std::string_view str, const F &f, rpc_result &ret) { + template + asio::awaitable handle_more_args(std::string_view str, const F &f, + rpc_result &ret, Self *self) { rpc_service::msgpack_codec codec; - auto tp = codec.unpack(str); + auto tp = codec.unpack(str); if constexpr (std::is_void_v) { - std::apply(f, tp); + if constexpr (std::is_void_v) { + if constexpr (is_awaitable_v) { + co_await std::apply(f, tp); + } else { + std::apply(f, tp); + } + } else { + if constexpr (is_awaitable_v) { + co_await std::apply( + [self, &f](auto &&...args) { + return (*self.*f)(std::forward(args)...); + }, + tp); + } else { + std::apply( + [self, &f](auto &&...args) { + return (*self.*f)(std::forward(args)...); + }, + tp); + } + } } else { - ret = rpc_service::msgpack_codec::pack_args(std::apply(f, tp)); + if constexpr (std::is_void_v) { + if constexpr (is_awaitable_v) { + ret = + rpc_service::msgpack_codec::pack_args(co_await std::apply(f, tp)); + } else { + ret = rpc_service::msgpack_codec::pack_args(std::apply(f, tp)); + } + } else { + if constexpr (is_awaitable_v) { + ret = rpc_service::msgpack_codec::pack_args(co_await std::apply( + [self, &f](auto &&...args) { + return (*self.*f)(std::forward(args)...); + }, + tp)); + } else { + ret = rpc_service::msgpack_codec::pack_args(std::apply( + [self, &f](auto &&...args) { + return (*self.*f)(std::forward(args)...); + }, + tp)); + } + } } } - template - void handle_more_args(std::string_view str, const F &f, Self *self, - rpc_result &ret) { - rpc_service::msgpack_codec codec; - auto tp = codec.unpack(str); - - if constexpr (std::is_void_v) { - std::apply( - [self, &f](auto &&...args) { - return (*self.*f)(std::forward(args)...); - }, - tp); - } else { - ret = rpc_service::msgpack_codec::pack_args(std::apply( - [self, &f](auto &&...args) { - return (*self.*f)(std::forward(args)...); - }, - tp)); - } - } - - std::unordered_map> + std::unordered_map( + std::string_view, rpc_result &)>> map_invokers_; std::unordered_map key2func_name_; }; diff --git a/tests/test_rest_rpc1.cpp b/tests/test_rest_rpc1.cpp index 0b38006..f8babf6 100644 --- a/tests/test_rest_rpc1.cpp +++ b/tests/test_rest_rpc1.cpp @@ -16,10 +16,24 @@ struct dummy { std::string echo(std::string val) { return val; } int round1(int i) { return i; } + asio::awaitable no_arg_coro() { + std::cout << "no args\n"; + co_return; + } + + asio::awaitable no_arg_coro1() { + std::cout << "no args\n"; + co_return "test"; + } + + asio::awaitable echo_coro(std::string str) { co_return str; } + asio::awaitable add_coro(int a, int b) { co_return a + b; } }; int add(int a, int b) { return a + b; } +asio::awaitable add_coro(int a, int b) { co_return a + b; } + void foo(std::string str) { std::cout << str << "\n"; } int round1(int i) { return i; } @@ -31,22 +45,25 @@ std::string_view delay_response(std::string_view str) { // set_delay before response in another thread ctx.set_delay(true); - std::thread thd([ctx=std::move(ctx)]() mutable { + std::thread thd([ctx = std::move(ctx)]() mutable { std::this_thread::sleep_for(std::chrono::seconds(2)); -// auto ec = ctx.sync_response("it is from a detached thread");//TODO: add a safe version -// if (ec) { -// REST_LOG_ERROR << "response error: " << ec.message(); -// } + // auto ec = ctx.sync_response("it is from a detached thread"); + // if (ec) { + // REST_LOG_ERROR << "response error: " << ec.message(); + // } auto executor = ctx.get_executor(); - auto coro = [ctx = std::move(ctx)]() mutable ->asio::awaitable { - auto ec = co_await ctx.response("test");//TODO: avoid duplicate response + auto coro = [ctx = std::move(ctx)]() mutable -> asio::awaitable { + auto ec = co_await ctx.response("test"); if (ec) { REST_LOG_ERROR << "response error: " << ec.message(); } + ec = co_await ctx.response("test"); + REST_LOG_ERROR << ec.message(); + CHECK(ec); }; - + async_start(executor, std::move(coro)); -// asio::co_spawn(executor, std::move(coro), asio::detached); + // asio::co_spawn(executor, std::move(coro), asio::detached); }); thd.detach(); @@ -61,53 +78,99 @@ asio::awaitable echo_coro(std::string str) { co_return str; } void no_arg() { std::cout << "no args\n"; } -TEST_CASE("test router") { +asio::awaitable no_arg_coro() { + std::cout << "no args\n"; + co_return; +} + +asio::awaitable no_arg_coro1() { + std::cout << "no args\n"; + co_return "test"; +} + +// TODO: handle connection lifetime, client pool, pub/sub +asio::awaitable test_router() { rpc_router router; router.register_handler(); router.register_handler(); router.register_handler(); router.register_handler(); - -// router.register_handler(); + + router.register_handler(); + router.register_handler(); + router.register_handler(); + router.register_handler(); + + { + auto ret = co_await router.route(get_key(), ""); + CHECK(ret.ec == rpc_errc::ok); + + auto ret1 = co_await router.route(get_key(), ""); + CHECK(ret1.ec == rpc_errc::ok); + + auto ret2 = co_await router.route(get_key(), "test"); + CHECK(ret2.ec == rpc_errc::ok); + } dummy d{}; router.register_handler<&dummy::add>(&d); router.register_handler<&dummy::foo>(&d); router.register_handler<&dummy::round1>(&d); router.register_handler<&dummy::echo>(&d); + router.register_handler<&dummy::no_arg_coro>(&d); + router.register_handler<&dummy::no_arg_coro1>(&d); + router.register_handler<&dummy::echo_coro>(&d); + router.register_handler<&dummy::add_coro>(&d); + { + auto ret = co_await router.route(get_key<&dummy::no_arg_coro>(), ""); + CHECK(ret.ec == rpc_errc::ok); + auto ret1 = co_await router.route(get_key<&dummy::no_arg_coro1>(), ""); + CHECK(ret1.ec == rpc_errc::ok); + auto ret2 = co_await router.route(get_key<&dummy::echo_coro>(), "test"); + CHECK(ret2.ec == rpc_errc::ok); + } rpc_service::msgpack_codec codec; + { auto s = codec.pack_args(1); auto s1 = codec.pack_args("test"); - auto r = router.route(get_key(), s); - auto r1 = router.route(get_key(), s1); + auto r = co_await router.route(get_key(), s); + auto r1 = co_await router.route(get_key(), s1); - auto r2 = router.route(get_key<&dummy::round1>(), s); - auto r3 = router.route(get_key<&dummy::echo>(), s1); + auto r2 = co_await router.route(get_key<&dummy::round1>(), s); + auto r3 = co_await router.route(get_key<&dummy::echo>(), s1); std::cout << "\n"; } auto args = codec.pack_args(1, 2); std::string_view str(args.data(), args.size()); + { + auto r1 = co_await router.route(get_key(), str); + auto r2 = co_await router.route(get_key<&dummy::add_coro>(), str); + std::cout << "\n"; + } + auto args1 = codec.pack_args("it is a test"); std::string_view str1(args1.data(), args1.size()); { - auto result = router.route(get_key<&dummy::add>(), str); + auto result = co_await router.route(get_key<&dummy::add>(), str); auto r = codec.unpack(result.result); - auto result1 = router.route(get_key<&dummy::foo>(), str1); + auto result1 = co_await router.route(get_key<&dummy::foo>(), str1); CHECK(r == 3); CHECK(result1.ec == rpc_errc::ok); } - auto result = router.route(get_key(), str); + auto result = co_await router.route(get_key(), str); CHECK(result.ec == rpc_errc::ok); - auto result1 = router.route(get_key(), str1); + auto result1 = co_await router.route(get_key(), str1); CHECK(result1.ec == rpc_errc::ok); } +TEST_CASE("test router") { sync_wait(get_global_executor(), test_router()); } + asio::awaitable void_returning_coroutine() { auto executor = co_await asio::this_coro::executor; asio::io_context &ioc = static_cast(executor.context()); @@ -153,7 +216,8 @@ TEST_CASE("test server start") { // auto future1 = asio::co_spawn(cl.get_executor(), cl.call(1, 2), // asio::use_future); auto result = future1.get(); { - auto result = sync_wait(cl.get_executor(), cl.call_for(std::chrono::minutes(2), 1, 2)); + auto result = sync_wait(cl.get_executor(), + cl.call_for(std::chrono::minutes(2), 1, 2)); CHECK(result.ec == rpc_errc::ok); } { @@ -175,7 +239,6 @@ TEST_CASE("test server start") { auto result2 = sync_wait(cl.get_executor(), cl.call(1)); CHECK(result2.ec == rpc_errc::ok); } - ec = server.async_start(); CHECK(!ec);