saberbud/rest_rpc
1
0
Fork 0
mirror of https://github.com/qicosmos/rest_rpc.git synced 2026-08-29 08:34:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

simplify

Browse Source
This commit is contained in:
qicosmos
2025-09-24 08:29:49 +08:00
parent f02df7ffa4
commit cad26c4c97
6 changed files with 227 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/rest_rpc/asio_util.hpp
+13 -9
View File
@@ -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<decltype(coro)>::value_type;
template<typename Coro>
inline auto async_start(auto executor, Coro &&coro) {
using R = typename std::remove_cvref_t<std::invoke_result_t<Coro>>::value_type;
static_assert(std::is_void_v<R>);
asio::co_spawn(executor, std::move(coro), asio::detached);
asio::co_spawn(executor, std::forward<Coro>(coro), asio::detached);
}
inline auto async_future(auto executor, auto &&coro) {
return asio::co_spawn(executor, std::move(coro), asio::use_future);
template<typename Coro>
inline auto async_future(auto executor, Coro &&coro) {
return asio::co_spawn(executor, std::forward<Coro>(coro), asio::use_future);
}
inline auto sync_wait(auto executor, auto &&coro) {
return async_future(executor, std::move(coro)).get();
template<typename Coro>
inline auto sync_wait(auto executor, Coro &&coro) {
return async_future(executor, std::forward<Coro>(coro)).get();
}
inline auto async_start(auto executor, auto &&coro, auto callback) {
include/rest_rpc/client.hpp
+10 -6
View File
@@ -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<decltype(func)>::return_type;
auto r = co_await(watchdog(duration) ||
call_impl<func>(std::forward<Args>(args)...));
auto r = co_await (watchdog(duration) ||
call_impl<func>(std::forward<Args>(args)...));
if (r.index() == 0) {
co_return call_result<R>{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<R>) {
result.value = codec.unpack<R>(
std::string_view(body_.data() + 1, resp_header.body_len - 1));
include/rest_rpc/rest_rpc_protocol.hpp
+3
View File
@@ -4,6 +4,8 @@
#include <winsock2.h>
#else
#include <arpa/inet.h>
#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;
include/rest_rpc/rpc_connection.hpp
+60 -23
View File
@@ -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<rpc_connection> 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<rpc_connection> 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 <typename T> void response(T &&t) {}
template <auto func, typename... Args>
asio::awaitable<std::error_code> response(Args &&...args);
template <auto func, typename... Args> std::error_code sync_response(Args &&...args);
private:
std::weak_ptr<rpc_connection> weak_;
inline static bool delay_ = false;
rpc_context() = default;
std::shared_ptr<rpc_connection> conn_;
bool delay_ = false;
};
class rpc_connection : public std::enable_shared_from_this<rpc_connection> {
@@ -40,7 +46,7 @@ public:
asio::awaitable<void> 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<asio::const_buffer> 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<std::error_code> 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<asio::const_buffer> 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 <auto func, typename... Args>
asio::awaitable<std::error_code> rpc_context::response(Args &&...args) {
using args_tuple = typename function_traits<decltype(func)>::return_type;
static_assert(std::is_constructible_v<args_tuple, Args...>,
"rpc function return type and response arguments are not match");
rpc_service::msgpack_codec codec;
rpc_result result(codec.pack_args(std::forward<Args>(args)...));
co_return co_await conn_->response(result);
}
template <auto func, typename... Args>
std::error_code rpc_context::sync_response(Args &&...args) {
return sync_wait(conn_->get_executor(), response<func>(std::forward<Args>(args)...));
}
auto rpc_context::get_executor() {
return conn_->get_executor();
}
} // namespace rest_rpc
include/rest_rpc/rpc_router.hpp
+96 -66
View File
@@ -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 <typename Function>
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<Function>::tuple_type;
using R = typename function_traits<Function>::return_type;
try {
if constexpr (std::tuple_size_v<args_tuple> == 0) {
if constexpr (std::is_void_v<R>) {
f();
} else {
auto r = f();
ret.result = rpc_service::msgpack_codec::pack_to_string(r);
}
handle_zero_arg<R>(f, ret);
} else {
rpc_service::msgpack_codec codec;
using first_t = std::tuple_element_t<0, args_tuple>;
if constexpr (std::tuple_size_v<args_tuple> == 1 &&
util::is_basic_v<first_t>) {
if constexpr (std::is_void_v<R>) {
f(codec.unpack<first_t>(str));
} else {
if constexpr (std::is_same_v<std::string_view, R>) {
ret.view = rpc_service::msgpack_codec::pack_args(
f(codec.unpack<first_t>(str)));
} else {
ret.result = rpc_service::msgpack_codec::pack_args(
f(codec.unpack<first_t>(str)));
}
}
handle_one_arg<R, first_t>(str, f, ret);
} else {
auto tp = codec.unpack<args_tuple>(str);
if constexpr (std::is_void_v<R>) {
std::apply(f, tp);
} else {
auto r = std::apply(f, tp);
ret.result = rpc_service::msgpack_codec::pack_to_string(r);
}
handle_more_args<R, args_tuple>(str, f, ret);
}
}
} catch (std::invalid_argument &e) {
@@ -150,54 +140,22 @@ private:
template <typename Function, typename Self>
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<Function>::tuple_type;
using R = typename function_traits<Function>::return_type;
rpc_service::msgpack_codec codec;
try {
if constexpr (std::tuple_size_v<args_tuple> == 0) {
if constexpr (std::is_void_v<R>) {
(*self.*f)();
} else {
auto r = (*self.*f)();
ret.result = rpc_service::msgpack_codec::pack_to_string(r);
}
handle_zero_arg<R>(f, self, ret);
} else {
using first_t = std::tuple_element_t<0, args_tuple>;
if constexpr (std::tuple_size_v<args_tuple> == 1 &&
util::is_basic_v<first_t>) {
if constexpr (std::is_void_v<R>) {
(*self.*f)(codec.unpack<first_t>(str));
} else {
if constexpr (std::is_same_v<std::string_view, R>) {
ret.view = rpc_service::msgpack_codec::pack_args(
(*self.*f)(codec.unpack<first_t>(str)));
} else {
ret.result = rpc_service::msgpack_codec::pack_args(
(*self.*f)(codec.unpack<first_t>(str)));
}
}
handle_one_arg<R, first_t>(str, f, self, ret);
} else {
auto tp = codec.unpack<args_tuple>(str);
if constexpr (std::is_void_v<R>) {
std::apply(
[self, &f](auto &&...args) {
return (*self.*f)(std::forward<decltype(args)>(args)...);
},
tp);
} else {
auto r = std::apply(
[self, &f](auto &&...args) {
return (*self.*f)(std::forward<decltype(args)>(args)...);
},
tp);
ret.result = rpc_service::msgpack_codec::pack_to_string(r);
}
handle_more_args<R, args_tuple>(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 <typename R, typename F>
void handle_zero_arg(const F &f, rpc_result &ret) {
if constexpr (std::is_void_v<R>) {
f();
} else {
ret = rpc_service::msgpack_codec::pack_args(f());
}
}
template <typename R, typename F, typename Self>
void handle_zero_arg(const F &f, Self *self, rpc_result &ret) {
if constexpr (std::is_void_v<R>) {
(*self.*f)();
} else {
ret = rpc_service::msgpack_codec::pack_args((*self.*f)());
}
}
template <typename R, typename Arg, typename F>
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<R>) {
f(codec.unpack<Arg>(str));
} else {
ret = rpc_service::msgpack_codec::pack_args(f(codec.unpack<Arg>(str)));
}
}
template <typename R, typename Arg, typename F, typename Self>
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<R>) {
(*self.*f)(codec.unpack<Arg>(str));
} else {
ret = rpc_service::msgpack_codec::pack_args(
(*self.*f)(codec.unpack<Arg>(str)));
}
}
template <typename R, typename args_tuple, typename F>
void handle_more_args(std::string_view str, const F &f, rpc_result &ret) {
rpc_service::msgpack_codec codec;
auto tp = codec.unpack<args_tuple>(str);
if constexpr (std::is_void_v<R>) {
std::apply(f, tp);
} else {
ret = rpc_service::msgpack_codec::pack_args(std::apply(f, tp));
}
}
template <typename R, typename args_tuple, typename F, typename Self>
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<args_tuple>(str);
if constexpr (std::is_void_v<R>) {
std::apply(
[self, &f](auto &&...args) {
return (*self.*f)(std::forward<decltype(args)>(args)...);
},
tp);
} else {
ret = rpc_service::msgpack_codec::pack_args(std::apply(
[self, &f](auto &&...args) {
return (*self.*f)(std::forward<decltype(args)>(args)...);
},
tp));
}
}
std::unordered_map<uint32_t,
std::function<void(std::string_view, rpc_result &)>>
map_invokers_;