Files
qing 81ba942b8e feat(libmsg): extract pymsg-compatible messaging library as shared lib
- Add libmsg: C++20 header-only pymsg protocol implementation
  - Msg.hpp: pmsg_send/pmsg_recv (blocking + deadline modes) + MsgClient
  - MsgServer.hpp: multi-client TCP server with select event loop + thread pool
  - AsyncQueue<T>: generic async batch processing queue
  - RpcClient/RpcServer: JSON RPC convention layer (requires boost::json)
  - Msg_Basic.hpp/cpp: exception hierarchy, constants, version
- Fix thread safety: MsgClient _sockfd/_running use std::atomic with exchange()
- Fix deadline recv: client call() uses shared deadline for header+payload
- Fix graceful shutdown: thread pool drains queued tasks before exit
- Fix pmsg_send size check: enforce 64MiB limit (was 4GiB no-op check)
- Align AsyncQueue batch semantics with Python: batch_size is threshold
- Refactor gateway/LogC.hpp to use libmsg AsyncQueue<T>
- Update gateway Makefile to link -lus_msg, DbC.hpp to use <us/Msg.hpp>
- Add libmsg to _inst.sh build order
2026-08-06 23:14:13 +08:00

182 lines
7.1 KiB
C++
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#pragma once
/** RpcClient — pymsg 线协议之上的 JSON RPC 客户端(C++ 版,对应 Python 版 pymsg.rpc.RpcClient)
*
* 约定(与 Python/Go 客户端字节级兼容):
* 请求: {"_route": "<路由名>", ...业务字段}
* 响应: {"success": true, "data": ...}
* {"success": false, "error": "..."}
*
* 基于 boost::json 进行 JSON 序列化/反序列化,依赖 MsgClient 进行 TCP 通信。
*
* 注意:此头文件依赖 boost::json,核心 libmsg 库(Msg.hpp/MsgServer.hpp/AsyncQueue.hpp)
* 保持零外部依赖不变。使用 RpcClient 时需要链接 -lboost_json。
*
* Usage:
* #include <us/RpcClient.hpp>
*
* pmc::net::RpcClient rpc("127.0.0.1", 8005, "gateway");
* json::object params;
* params["action"] = "select";
* params["table"] = "Users";
* json::value data = rpc.call("execute", params); // 返回 data 字段
*
* // 启动等待模式(db 未启动时阻塞重试)
* pmc::net::RpcClient rpc("127.0.0.1", 8005, "gateway", 10.0, true);
*/
#include <us/Msg.hpp>
#include <cstddef>
#include <optional>
#include <stdexcept>
#include <string>
#include <boost/json.hpp>
namespace pmc {
namespace net {
namespace json = boost::json;
// ============================================================================
// RpcError — RPC 业务异常
// ============================================================================
/// 远端返回 success=false 的业务错误,或响应不符合 RPC 约定(非 JSON / 缺少 success 字段)。
/// 网络层异常(超时、连接拒绝等)仍抛 MsgTimeoutException / MsgConnectionRefusedException 等。
class RpcError : public MsgException {
public:
explicit RpcError(const std::string& msg) : MsgException(msg) {}
};
// ============================================================================
// RpcClient
// ============================================================================
/// JSON RPC 客户端(短连接,每次调用独立 TCP 连接)。
class RpcClient {
public:
/** 构造 RPC 客户端。
*
* @param addr 服务地址(通常 127.0.0.1)
* @param port 服务端口
* @param name 客户端名称(用于调试/日志)
* @param timeout 默认超时秒数
* @param retry_on_refused 默认是否在 ECONNREFUSED 时阻塞重试
* @param retry_interval 重试间隔秒数
*/
explicit RpcClient(const std::string& addr, int port,
const std::string& name = "rpc",
double timeout = 10.0,
bool retry_on_refused = false,
double retry_interval = 1.0)
: _addr(addr), _port(port), _name(name),
_timeout(timeout),
_retry_on_refused(retry_on_refused),
_retry_interval(retry_interval) {}
// ---- RPC 调用 ----------------------------------------------------------
/** 调用指定路由,返回响应中的 data 字段。
*
* @param route 路由名(自动注入为 _route 字段)
* @param params 业务参数(_route 会被覆盖)
* @param timeout 本次调用超时(nullopt = 使用构造时默认值)
* @param retry_on_refused 本次调用是否重试连接拒绝(nullopt = 使用构造时默认值)
* @return 服务端返回的 data 字段值(可能为 null)
* @throws RpcError 服务端返回 success=false,或响应非法
* @throws MsgTimeoutException 网络超时
* @throws MsgConnectionRefusedException 目标不可达且未重试
* @throws MsgConnectionException 连接中断
*/
json::value call(const std::string& route,
const json::object& params = {},
std::optional<double> timeout = std::nullopt,
std::optional<bool> retry_on_refused = std::nullopt,
std::optional<double> retry_interval = std::nullopt) {
json::object req = params;
req["_route"] = route;
double t = timeout.value_or(_timeout);
bool retry = retry_on_refused.value_or(_retry_on_refused);
double interval = retry_interval.value_or(_retry_interval);
MsgClient client(_addr, _port, _name);
std::string resp_str = client.call(json::serialize(req), t, retry, interval);
json::value resp;
try {
resp = json::parse(resp_str);
} catch (const std::exception& e) {
throw RpcError("响应不是合法 JSON: " + std::string(e.what()));
}
if (!resp.is_object()) {
throw RpcError("非法 RPC 响应(不是 object): " + resp_str.substr(0, 200));
}
const json::object& obj = resp.as_object();
if (!obj.contains("success")) {
throw RpcError("非法 RPC 响应(缺少 success 字段): " + resp_str.substr(0, 200));
}
if (!obj.at("success").as_bool()) {
std::string err = "未知错误";
if (auto it = obj.find("error"); it != obj.end() && it->value().is_string()) {
err = it->value().as_string().c_str();
}
throw RpcError(err);
}
if (auto it = obj.find("data"); it != obj.end()) {
return it->value();
}
return nullptr;
}
/** 调用指定路由,返回完整响应对象(包含 success/data/error),不抛 RpcError。
* 适用于需要自行检查 success 字段或访问额外字段的场景。
*/
json::object call_raw(const std::string& route,
const json::object& params = {},
std::optional<double> timeout = std::nullopt,
std::optional<bool> retry_on_refused = std::nullopt,
std::optional<double> retry_interval = std::nullopt) {
json::object req = params;
req["_route"] = route;
double t = timeout.value_or(_timeout);
bool retry = retry_on_refused.value_or(_retry_on_refused);
double interval = retry_interval.value_or(_retry_interval);
MsgClient client(_addr, _port, _name);
std::string resp_str = client.call(json::serialize(req), t, retry, interval);
json::value resp;
try {
resp = json::parse(resp_str);
} catch (const std::exception& e) {
throw RpcError("响应不是合法 JSON: " + std::string(e.what()));
}
if (!resp.is_object()) {
throw RpcError("非法 RPC 响应(不是 object): " + resp_str.substr(0, 200));
}
return resp.as_object();
}
// ---- 访问器 ------------------------------------------------------------
const std::string& addr() const noexcept { return _addr; }
int port() const noexcept { return _port; }
const std::string& name() const noexcept { return _name; }
private:
std::string _addr;
int _port;
std::string _name;
double _timeout;
bool _retry_on_refused;
double _retry_interval;
};
} // namespace net
} // namespace pmc