From cad26c4c97008063a0b49bf8bd7ae1d472e9183b Mon Sep 17 00:00:00 2001 From: qicosmos Date: Wed, 24 Sep 2025 08:29:49 +0800 Subject: [PATCH] simplify --- include/rest_rpc/asio_util.hpp | 22 ++-- include/rest_rpc/client.hpp | 16 ++- include/rest_rpc/rest_rpc_protocol.hpp | 3 + include/rest_rpc/rpc_connection.hpp | 83 +++++++++---- include/rest_rpc/rpc_router.hpp | 162 +++++++++++++++---------- tests/test_rest_rpc1.cpp | 49 +++++++- 6 files changed, 227 insertions(+), 108 deletions(-) diff --git a/include/rest_rpc/asio_util.hpp b/include/rest_rpc/asio_util.hpp index d8cdd2a..c983a39 100644 --- a/include/rest_rpc/asio_util.hpp +++ b/include/rest_rpc/asio_util.hpp @@ -2,18 +2,22 @@ #include "use_asio.hpp" namespace rest_rpc { -inline auto async_start(auto executor, auto &&coro) { - using R = typename std::remove_cvref_t::value_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::move(coro), asio::detached); + asio::co_spawn(executor, std::forward(coro), asio::detached); } - -inline auto async_future(auto executor, auto &&coro) { - return asio::co_spawn(executor, std::move(coro), asio::use_future); + + template +inline auto async_future(auto executor, Coro &&coro) { + return asio::co_spawn(executor, std::forward(coro), asio::use_future); } - -inline auto sync_wait(auto executor, auto &&coro) { - return async_future(executor, std::move(coro)).get(); + + template +inline auto sync_wait(auto executor, Coro &&coro) { + return async_future(executor, std::forward(coro)).get(); } inline auto async_start(auto executor, auto &&coro, auto callback) { diff --git a/include/rest_rpc/client.hpp b/include/rest_rpc/client.hpp index 249cfe7..0f85391 100644 --- a/include/rest_rpc/client.hpp +++ b/include/rest_rpc/client.hpp @@ -33,9 +33,9 @@ public: std::chrono::steady_clock::duration duration = std::chrono::seconds(5)) { asio::ip::tcp::resolver resolver(socket_.get_executor()); - auto r = co_await(watchdog(duration) || - resolver.async_resolve( - host, port, asio::as_tuple(asio::use_awaitable))); + auto r = co_await (watchdog(duration) || + resolver.async_resolve( + host, port, asio::as_tuple(asio::use_awaitable))); if (r.index() == 0) { REST_LOG_ERROR << "resolve timeout"; co_return make_error_code(rpc_errc::resolve_timeout); @@ -54,7 +54,7 @@ public: } auto endpoint = it->endpoint(); - auto conn_r = co_await( + auto conn_r = co_await ( watchdog(duration) || socket_.async_connect(endpoint, asio::as_tuple(asio::use_awaitable))); if (conn_r.index() == 0) { @@ -105,8 +105,8 @@ public: "called rpc function and arguments are not match"); using R = typename function_traits::return_type; - auto r = co_await(watchdog(duration) || - call_impl(std::forward(args)...)); + auto r = co_await (watchdog(duration) || + call_impl(std::forward(args)...)); if (r.index() == 0) { co_return call_result{rpc_errc::request_timeout}; } @@ -177,6 +177,10 @@ private: co_return result; } result.ec = (rpc_errc)body_[0]; + if (resp_header.body_len > 0) { + auto view = std::string_view(body_.data() + 1, resp_header.body_len - 1); + REST_LOG_INFO << view; + } if constexpr (!std::is_void_v) { result.value = codec.unpack( std::string_view(body_.data() + 1, resp_header.body_len - 1)); diff --git a/include/rest_rpc/rest_rpc_protocol.hpp b/include/rest_rpc/rest_rpc_protocol.hpp index a7004f0..453923e 100644 --- a/include/rest_rpc/rest_rpc_protocol.hpp +++ b/include/rest_rpc/rest_rpc_protocol.hpp @@ -4,6 +4,8 @@ #include #else #include +#ifdef __APPLE__ +#else inline uint64_t htonll(uint64_t value) { return ((uint64_t)htonl(value & 0xFFFFFFFF) << 32) | htonl(value >> 32); } @@ -12,6 +14,7 @@ inline uint64_t ntohll(uint64_t value) { return ((uint64_t)ntohl(value & 0xFFFFFFFF) << 32) | ntohl(value >> 32); } #endif +#endif namespace rest_rpc { inline constexpr uint8_t REST_MAGIC_NUM = 39; diff --git a/include/rest_rpc/rpc_connection.hpp b/include/rest_rpc/rpc_connection.hpp index 3efbeaa..dff2164 100644 --- a/include/rest_rpc/rpc_connection.hpp +++ b/include/rest_rpc/rpc_connection.hpp @@ -12,23 +12,29 @@ class rpc_context { public: static auto &context() { thread_local rpc_context instance; - delay_ = true; return instance; } - void set_connection(std::shared_ptr conn) { weak_ = conn; } + rpc_context(rpc_context&& o) : conn_(std::move(o.conn_)), delay_(o.delay_) {} + rpc_context(const rpc_context& o) = delete; - auto get_connection() { return weak_; } + void set_connection(std::shared_ptr conn) { conn_ = conn; } - static bool delay() { return delay_; } + bool delay() { return delay_; } - static bool set_delay(bool r) { delay_ = r; } + void set_delay(bool r) { context().delay_ = r; } + + auto get_executor(); - template void response(T &&t) {} + template + asio::awaitable response(Args &&...args); + + template std::error_code sync_response(Args &&...args); private: - std::weak_ptr weak_; - inline static bool delay_ = false; + rpc_context() = default; + std::shared_ptr conn_; + bool delay_ = false; }; class rpc_connection : public std::enable_shared_from_this { @@ -40,7 +46,7 @@ public: asio::awaitable start() { rest_rpc_header header; - + auto self = this->shared_from_this(); while (true) { std::error_code ec; size_t size; @@ -73,32 +79,43 @@ public: } // route + rpc_context::context().set_connection(self); auto result = router_.route(header.function_id, body_); - rest_rpc_header resp_header{}; - resp_header.magic = 39; - resp_header.body_len = result.size() + 1; - if (cross_ending_) { - prepare_for_send(resp_header); + bool delay = rpc_context::context().delay(); + if (delay) { + rpc_context::context().set_delay(false); + continue; } - std::vector buffers; - buffers.reserve(3); - buffers.push_back(asio::buffer(&resp_header, sizeof(rest_rpc_header))); - buffers.push_back(asio::buffer(&result.ec, 1)); - if (!result.empty()) - buffers.push_back(asio::buffer(result.data())); - std::tie(ec, size) = co_await asio::async_write( - socket_, buffers, asio::as_tuple(asio::use_awaitable)); + ec = co_await response(result); if (ec) { REST_LOG_WARNING << "write error: " << ec.message(); break; } } + } - co_return; + asio::awaitable response(const rpc_result &result) { + rest_rpc_header resp_header{}; + resp_header.magic = 39; + resp_header.body_len = result.size() + 1; + if (cross_ending_) { + prepare_for_send(resp_header); + } + std::vector buffers; + buffers.reserve(3); + buffers.push_back(asio::buffer(&resp_header, sizeof(rest_rpc_header))); + buffers.push_back(asio::buffer(&result.ec, 1)); + if (!result.empty()) + buffers.push_back(asio::buffer(result.data())); + + auto [ec, size] = co_await asio::async_write( + socket_, buffers, asio::as_tuple(asio::use_awaitable)); + co_return ec; } uint64_t id() const { return conn_id_; } + auto get_executor() { return socket_.get_executor(); } private: tcp_socket socket_; @@ -107,4 +124,24 @@ private: rpc_router &router_; bool cross_ending_; }; + +// zero or one arguments +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"); + rpc_service::msgpack_codec codec; + rpc_result result(codec.pack_args(std::forward(args)...)); + 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 9a3ca27..21fc3cb 100644 --- a/include/rest_rpc/rpc_router.hpp +++ b/include/rest_rpc/rpc_router.hpp @@ -11,13 +11,24 @@ namespace rest_rpc { struct rpc_result { + rpc_result(std::string str) : result(std::move(str)) {} + rpc_result(std::string_view str) : view(str) {} + rpc_result &operator=(std::string str) { + result = std::move(str); + return *this; + } + rpc_result &operator=(std::string_view str) { + result = str; + return *this; + } + rpc_result() = default; rpc_errc ec = rpc_errc::ok; std::string result; std::string_view view; - bool empty() { return result.empty() && view.empty(); } - size_t size() { return result.empty() ? view.size() : result.size(); } + bool empty() const { return result.empty() && view.empty(); } + size_t size() const { return result.empty() ? view.size() : result.size(); } - std::string_view data() { return result.empty() ? view : result; } + std::string_view data() const { return result.empty() ? view : result; } }; class rpc_router { @@ -99,43 +110,22 @@ private: template void register_nonmember_func(uint32_t key, Function f) { - this->map_invokers_[key] = [f = std::move(f)](std::string_view str, + 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; try { if constexpr (std::tuple_size_v == 0) { - if constexpr (std::is_void_v) { - f(); - } else { - auto r = f(); - ret.result = rpc_service::msgpack_codec::pack_to_string(r); - } + handle_zero_arg(f, ret); } else { - rpc_service::msgpack_codec codec; using first_t = std::tuple_element_t<0, args_tuple>; if constexpr (std::tuple_size_v == 1 && util::is_basic_v) { - if constexpr (std::is_void_v) { - f(codec.unpack(str)); - } else { - if constexpr (std::is_same_v) { - ret.view = rpc_service::msgpack_codec::pack_args( - f(codec.unpack(str))); - } else { - ret.result = rpc_service::msgpack_codec::pack_args( - f(codec.unpack(str))); - } - } + handle_one_arg(str, f, ret); } else { - auto tp = codec.unpack(str); - if constexpr (std::is_void_v) { - std::apply(f, tp); - } else { - auto r = std::apply(f, tp); - ret.result = rpc_service::msgpack_codec::pack_to_string(r); - } + handle_more_args(str, f, ret); } } } catch (std::invalid_argument &e) { @@ -150,54 +140,22 @@ private: template void register_member_func(uint32_t key, const Function &f, Self *self) { - this->map_invokers_[key] = [f, self](std::string_view str, - rpc_result &ret) { + this->map_invokers_[key] = [this, f, self](std::string_view str, + rpc_result &ret) { using args_tuple = typename function_traits::tuple_type; using R = typename function_traits::return_type; - rpc_service::msgpack_codec codec; try { if constexpr (std::tuple_size_v == 0) { - if constexpr (std::is_void_v) { - (*self.*f)(); - } else { - auto r = (*self.*f)(); - ret.result = rpc_service::msgpack_codec::pack_to_string(r); - } + handle_zero_arg(f, self, ret); } else { using first_t = std::tuple_element_t<0, args_tuple>; if constexpr (std::tuple_size_v == 1 && util::is_basic_v) { - if constexpr (std::is_void_v) { - (*self.*f)(codec.unpack(str)); - } else { - if constexpr (std::is_same_v) { - ret.view = rpc_service::msgpack_codec::pack_args( - (*self.*f)(codec.unpack(str))); - } else { - ret.result = rpc_service::msgpack_codec::pack_args( - (*self.*f)(codec.unpack(str))); - } - } + handle_one_arg(str, f, self, ret); } else { - 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 { - auto r = std::apply( - [self, &f](auto &&...args) { - return (*self.*f)(std::forward(args)...); - }, - tp); - ret.result = rpc_service::msgpack_codec::pack_to_string(r); - } + handle_more_args(str, f, self, ret); } } - } catch (std::invalid_argument &e) { ret.ec = rpc_errc::invalid_argument; ret.result = e.what(); @@ -208,6 +166,78 @@ 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)(); + } 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))); + } + } + + template + void handle_one_arg(std::string_view str, const F &f, Self *self, + rpc_result &ret) { + rpc_service::msgpack_codec codec; + if constexpr (std::is_void_v) { + (*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) { + rpc_service::msgpack_codec codec; + auto tp = codec.unpack(str); + if constexpr (std::is_void_v) { + std::apply(f, tp); + } else { + ret = rpc_service::msgpack_codec::pack_args(std::apply(f, 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> map_invokers_; diff --git a/tests/test_rest_rpc1.cpp b/tests/test_rest_rpc1.cpp index ffa4f28..0b38006 100644 --- a/tests/test_rest_rpc1.cpp +++ b/tests/test_rest_rpc1.cpp @@ -26,8 +26,39 @@ int round1(int i) { return i; } std::string_view echo_sv(std::string_view str) { return str; } +std::string_view delay_response(std::string_view str) { + auto &ctx = rpc_context::context(); + // 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)); +// 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 executor = ctx.get_executor(); + auto coro = [ctx = std::move(ctx)]() mutable ->asio::awaitable { + auto ec = co_await ctx.response("test");//TODO: avoid duplicate response + if (ec) { + REST_LOG_ERROR << "response error: " << ec.message(); + } + }; + + async_start(executor, std::move(coro)); +// asio::co_spawn(executor, std::move(coro), asio::detached); + }); + thd.detach(); + + // this return value is meaningless, because it will response later, the + // return type is important for client, so just return an empty value here. + return ""; +} + std::string echo(std::string str) { return str; } +asio::awaitable echo_coro(std::string str) { co_return str; } + void no_arg() { std::cout << "no args\n"; } TEST_CASE("test router") { @@ -36,6 +67,8 @@ TEST_CASE("test router") { router.register_handler(); router.register_handler(); router.register_handler(); + +// router.register_handler(); dummy d{}; router.register_handler<&dummy::add>(&d); @@ -62,7 +95,7 @@ TEST_CASE("test router") { { auto result = router.route(get_key<&dummy::add>(), str); - auto r = codec.unpack(result.result.data(), result.result.size()); + auto r = codec.unpack(result.result); auto result1 = router.route(get_key<&dummy::foo>(), str1); CHECK(r == 3); CHECK(result1.ec == rpc_errc::ok); @@ -90,6 +123,7 @@ TEST_CASE("test server start") { server.register_handler(); server.register_handler(); server.register_handler(); + server.register_handler(); auto ec = server.async_start(); CHECK(!ec); client cl; @@ -118,10 +152,19 @@ 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)); + CHECK(result.ec == rpc_errc::ok); + } { // auto result = sync_wait(cl.get_executor(), // cl.call_for(std::chrono::minutes(2))); // CHECK(result.ec==rpc_errc::ok); + auto result0 = + sync_wait(cl.get_executor(), + cl.call_for(std::chrono::minutes(2), "test")); + CHECK(result0.ec == rpc_errc::ok); + auto result = sync_wait(cl.get_executor(), cl.call_for(std::chrono::minutes(2), "test")); @@ -132,9 +175,7 @@ TEST_CASE("test server start") { auto result2 = sync_wait(cl.get_executor(), cl.call(1)); CHECK(result2.ec == rpc_errc::ok); } - - auto result = sync_wait(cl.get_executor(), cl.call(1, 2)); - CHECK(result.ec == rpc_errc::ok); + ec = server.async_start(); CHECK(!ec);