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

ending

Browse Source
This commit is contained in:
qicosmos
2025-09-22 19:33:53 +08:00
parent 8d0a65163c
commit 5e501503fb
6 changed files with 38 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/rest_rpc/client.hpp
+17 -11
View File
@@ -20,9 +20,7 @@ template <typename R> struct call_result {
R value;
};
template <> struct call_result<void> {
rpc_errc ec;
};
template <> struct call_result<void> { rpc_errc ec; };
class client {
public:
@@ -35,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);
@@ -56,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) {
@@ -107,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};
}
@@ -117,6 +115,8 @@ public:
void enable_tcp_no_delay(bool r) { tcp_no_delay_ = r; }
void enable_cross_ending(bool r) { cross_ending_ = r; }
private:
template <auto func, typename... Args>
asio::awaitable<
@@ -128,7 +128,9 @@ private:
rpc_service::msgpack_codec codec;
auto buf = codec.pack_args(std::forward<Args>(args)...);
header.body_len = buf.size();
prepare_for_send(header);
if (cross_ending_) {
prepare_for_send(header);
}
std::vector<asio::const_buffer> buffers;
buffers.reserve(2);
@@ -160,7 +162,10 @@ private:
result.ec = rpc_errc::protocol_error;
co_return result;
}
parse_recieved(resp_header);
if (cross_ending_) {
parse_recieved(resp_header);
}
detail::resize(body_, resp_header.body_len);
std::tie(ec, size) = co_await asio::async_read(
@@ -190,5 +195,6 @@ private:
tcp_socket socket_;
std::string body_;
bool tcp_no_delay_ = true;
bool cross_ending_ = false;
};
} // namespace rest_rpc
include/rest_rpc/rest_rpc_server.hpp
+5 -2
View File
@@ -70,6 +70,8 @@ public:
void enable_tcp_no_delay(bool r) { tcp_no_delay_ = r; }
void enable_cross_ending(bool r) { cross_ending_ = r; }
private:
std::error_code listen() {
using asio::ip::tcp;
@@ -132,8 +134,8 @@ private:
}
REST_LOG_INFO << "new connction comming...";
auto conn =
std::make_shared<rpc_connection>(std::move(socket), conn_id, router_);
auto conn = std::make_shared<rpc_connection>(std::move(socket), conn_id,
router_, cross_ending_);
conns_.emplace(conn_id++, conn);
co_spawn(socket.get_executor(), conn->start(), asio::detached);
}
@@ -176,5 +178,6 @@ private:
std::unordered_map<uint64_t, std::shared_ptr<rpc_connection>> conns_;
rpc_router router_;
bool tcp_no_delay_ = true;
bool cross_ending_ = false;
};
} // namespace rest_rpc
include/rest_rpc/rpc_connection.hpp
+12 -4
View File
@@ -8,8 +8,10 @@
namespace rest_rpc {
class rpc_connection {
public:
rpc_connection(tcp_socket socket, uint64_t conn_id, rpc_router &router)
: socket_(std::move(socket)), conn_id_(conn_id), router_(router) {}
rpc_connection(tcp_socket socket, uint64_t conn_id, rpc_router &router,
bool &cross_ending)
: socket_(std::move(socket)), conn_id_(conn_id), router_(router),
cross_ending_(cross_ending) {}
asio::awaitable<void> start() {
rest_rpc_header header;
@@ -25,7 +27,10 @@ public:
break;
}
parse_recieved(header);
if (cross_ending_) {
parse_recieved(header);
}
if (header.magic != REST_MAGIC_NUM) {
REST_LOG_ERROR << "protocol error";
break;
@@ -47,7 +52,9 @@ public:
rest_rpc_header resp_header{};
resp_header.magic = 39;
resp_header.body_len = result.size() + 1;
prepare_for_send(resp_header);
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)));
@@ -73,5 +80,6 @@ private:
uint64_t conn_id_;
std::string body_;
rpc_router &router_;
bool cross_ending_;
};
} // namespace rest_rpc
include/rest_rpc/rpc_router.hpp
+2 -2
View File
@@ -25,7 +25,7 @@ public:
template <typename Function, typename Self = void>
void register_handler(std::string_view name, const Function &f,
Self *self = nullptr) {
uint32_t key = MD5::MD5Hash32(name.data(), name.length());
uint32_t key = MD5::MD5Hash32(name.data(), (uint32_t)name.length());
register_handler_impl(key, name, f, self);
}
@@ -36,7 +36,7 @@ public:
}
void remove_handler(std::string_view name) {
uint32_t key = MD5::MD5Hash32(name.data(), name.length());
uint32_t key = MD5::MD5Hash32(name.data(), (uint32_t)name.length());
this->map_invokers_.erase(key);
key2func_name_.erase(key);
}
include/rest_rpc/util.hpp
+1 -1
View File
@@ -5,7 +5,7 @@
namespace rest_rpc {
template <auto func> constexpr uint32_t get_key() {
constexpr auto name = get_func_name<func>();
constexpr uint32_t key = MD5::MD5Hash32(name.data(), name.length());
constexpr uint32_t key = MD5::MD5Hash32(name.data(), (uint32_t)name.length());
return key;
}
} // namespace rest_rpc
tests/bench.cpp
+1 -1
View File
@@ -4,7 +4,7 @@
using namespace rest_rpc;
std::string address = "0.0.0.0:9004";
std::string address = "127.0.0.1:9004";
std::atomic<size_t> g_qps = 0;