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

pub sub

Browse Source
This commit is contained in:
qicosmos
2025-10-05 17:53:49 +08:00
parent 74b30ad302
commit 056e50c5e9
5 changed files with 109 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/rest_rpc/error_code.h
+3
View File
@@ -19,6 +19,7 @@ enum class rpc_errc : std::int8_t {
request_timeout,
protocol_error,
has_response,
duplicate_topic,
};
class rpc_error_category : public std::error_category {
@@ -57,6 +58,8 @@ public:
return "protocol error";
case rpc_errc::has_response:
return "has response, duplicate response is not allowed";
case rpc_errc::duplicate_topic:
return "duplicate topic";
default:
return "unknown error";
}
include/rest_rpc/rest_rpc_protocol.hpp
+2 -2
View File
@@ -6,7 +6,7 @@
#include <arpa/inet.h>
#endif
#if defined(__APPLE__) || defined(_WIN32)
#if defined(__APPLE__) || defined(_WIN32)
#else
inline uint64_t htonll(uint64_t value) {
return ((uint64_t)htonl(value & 0xFFFFFFFF) << 32) | htonl(value >> 32);
@@ -20,7 +20,7 @@ inline uint64_t ntohll(uint64_t value) {
namespace rest_rpc {
inline constexpr uint8_t REST_MAGIC_NUM = 39;
struct rest_rpc_header {
uint8_t magic;
uint8_t magic = REST_MAGIC_NUM;
uint8_t version;
uint8_t serialize_type;
uint8_t msg_type;
include/rest_rpc/rpc_client.hpp
+70 -13
View File
@@ -104,26 +104,56 @@ public:
static_assert(std::is_constructible_v<args_tuple, Args...>,
"called rpc function and arguments are not match");
rest_rpc_header header{};
header.function_id = get_key<func>();
using R = function_return_type_t<decltype(func)>;
auto r = co_await (watchdog(duration) ||
call_impl<func>(std::forward<Args>(args)...));
call_impl<R>(header, std::forward<Args>(args)...));
if (r.index() == 0) {
co_return call_result<R>{rpc_errc::request_timeout};
}
co_return std::get<1>(r);
}
template <typename R = void>
asio::awaitable<call_result<R>> subscribe(std::string_view topic) {
uint32_t topic_id = MD5::MD5Hash32(topic.data(), topic.size()); // topic id
bool b = false;
call_result<R> ret{};
auto it = sub_ops_.find(topic_id);
if (it == sub_ops_.end()) {
rest_rpc_header header{};
header.msg_type = 1; // pub/sub
header.function_id = topic_id;
auto [it, r] = sub_ops_.emplace(topic_id, sub_operation{});
if (!r) {
REST_LOG_ERROR << "subscribe duplicate topic";
co_return call_result<R>{rpc_errc::duplicate_topic};
}
std::tie(b, ret) = co_await (
asio::async_compose<decltype(asio::use_awaitable), void(bool)>(
std::ref(it->second), asio::use_awaitable) &&
call_impl<R>(header));
} else {
std::tie(b, ret) = co_await (
asio::async_compose<decltype(asio::use_awaitable), void(bool)>(
std::ref(it->second), asio::use_awaitable) &&
wait_response<R>());
}
co_return std::move(ret);
}
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<call_result<function_return_type_t<decltype(func)>>>
call_impl(Args &&...args) {
rest_rpc_header header{};
header.magic = 39;
header.function_id = get_key<func>();
template <typename R, typename... Args>
asio::awaitable<call_result<R>> call_impl(rest_rpc_header &header,
Args &&...args) {
rpc_service::msgpack_codec codec;
auto buf = codec.pack_args(std::forward<Args>(args)...);
header.body_len = buf.size();
@@ -138,7 +168,6 @@ private:
buffers.push_back(asio::buffer(buf.data(), buf.size()));
}
using R = function_return_type_t<decltype(func)>;
call_result<R> result{};
std::error_code ec;
size_t size;
@@ -149,6 +178,13 @@ private:
co_return result;
}
co_return co_await wait_response<R>();
}
template <typename R> asio::awaitable<call_result<R>> wait_response() {
call_result<R> result{};
std::error_code ec;
size_t size;
rest_rpc_header resp_header;
std::tie(ec, size) = co_await asio::async_read(
socket_, asio::buffer(&resp_header, sizeof(rest_rpc_header)),
@@ -176,16 +212,19 @@ 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>) {
rpc_service::msgpack_codec codec;
result.value = codec.unpack<R>(
std::string_view(body_.data() + 1, resp_header.body_len - 1));
}
co_return result;
if (resp_header.msg_type == 1) { // pubsub
if (auto it = sub_ops_.find(resp_header.function_id);
it != sub_ops_.end()) {
it->second.complete(true);
}
}
co_return std::move(result);
}
asio::awaitable<std::error_code> watchdog(auto duration) {
@@ -195,9 +234,27 @@ private:
co_return ec;
}
class sub_operation {
public:
template <typename Self> void operator()(Self &&self) {
using SelfType = std::decay_t<Self>;
auto shared_self = std::make_shared<SelfType>(std::move(self));
complete_handler_ = [shared_self](bool r) mutable {
shared_self->complete(r);
};
}
void complete(bool r) { complete_handler_(r); }
private:
std::function<void(bool)> complete_handler_;
};
tcp_socket socket_;
std::string body_;
bool tcp_no_delay_ = true;
bool cross_ending_ = false;
std::unordered_map<uint32_t, sub_operation> sub_ops_;
};
} // namespace rest_rpc
include/rest_rpc/rpc_connection.hpp
+13 -1
View File
@@ -76,6 +76,11 @@ public:
break;
}
if (header.msg_type == 1) { // pub sub
topic_id_ = header.function_id;
continue;
}
detail::resize(body_, header.body_len);
if (header.body_len > 0) {
@@ -105,9 +110,14 @@ public:
}
}
asio::awaitable<std::error_code> response(const rpc_result &result) {
asio::awaitable<std::error_code> response(const rpc_result &result,
uint32_t func_id = 0) {
rest_rpc_header resp_header{};
resp_header.magic = 39;
if (func_id != 0) {
resp_header.msg_type = 1;
resp_header.function_id = func_id;
}
resp_header.body_len = result.size() + 1;
if (cross_ending_) {
prepare_for_send(resp_header);
@@ -131,6 +141,7 @@ public:
uint64_t id() const { return conn_id_; }
auto get_executor() { return socket_.get_executor(); }
uint32_t topic_id() const { return topic_id_; }
void
set_quit_callback(std::function<void(const uint64_t &conn_id)> callback) {
@@ -178,6 +189,7 @@ private:
bool checkout_timeout_ = false;
rpc_router &router_;
bool cross_ending_;
uint32_t topic_id_;
};
// zero or one arguments
include/rest_rpc/rpc_server.hpp
+21
View File
@@ -102,6 +102,27 @@ public:
return conns_.size();
}
auto get_connections() {
std::scoped_lock lock(*conn_mtx_);
return conns_;
}
template <typename T>
asio::awaitable<void> publish(std::string_view topic, T &&t) {
auto id = MD5::MD5Hash32(topic.data(), topic.size());
auto conns = get_connections();
for (auto &[_, conn] : conns) {
if (conn->topic_id() == id) {
co_await conn->response(
rpc_service::msgpack_codec::pack_args(std::forward<T>(t)), id);
}
}
}
template <typename T> void sync_publish(std::string_view topic, T &&t) {
sync_wait(get_global_executor(), publish(topic, std::forward<T>(t)));
}
private:
std::error_code listen() {
using asio::ip::tcp;