363 lines
14 KiB
C++
363 lines
14 KiB
C++
#pragma warning(disable : 4251)
|
|
|
|
#include "rpc_server.h"
|
|
|
|
#include <chrono>
|
|
#include <condition_variable>
|
|
#include <filesystem>
|
|
#include <memory>
|
|
#include <mutex>
|
|
#include <queue>
|
|
#include <random>
|
|
#include <sstream>
|
|
#include <string>
|
|
#include <thread>
|
|
|
|
#include <magic_enum/magic_enum.hpp>
|
|
#include <nng/protocol/pair1/pair.h>
|
|
#include <nng/supplemental/util/platform.h>
|
|
|
|
#include "account_manager.h"
|
|
#include "chatroom_manager.h"
|
|
#include "contact_manager.h"
|
|
#include "database_executor.h"
|
|
#include "log.hpp"
|
|
#include "message_handler.h"
|
|
#include "message_sender.h"
|
|
#include "misc_manager.h"
|
|
#include "pb_types.h"
|
|
#include "pb_util.h"
|
|
#include "rpc_helper.h"
|
|
#include "spy.h"
|
|
#include "spy_types.h"
|
|
#include "util.h"
|
|
|
|
namespace fs = std::filesystem;
|
|
|
|
constexpr size_t DEFAULT_BUF_SIZE = 16 * 1024 * 1024;
|
|
|
|
std::unique_ptr<RpcServer, RpcServer::Deleter> RpcServer::instance_ = nullptr;
|
|
|
|
RpcServer &RpcServer::getInstance()
|
|
{
|
|
if (!instance_) {
|
|
instance_.reset(new RpcServer());
|
|
}
|
|
return *instance_;
|
|
}
|
|
|
|
void RpcServer::destroyInstance()
|
|
{
|
|
if (instance_) {
|
|
instance_->stop();
|
|
instance_.reset();
|
|
}
|
|
}
|
|
|
|
RpcServer::RpcServer(int port)
|
|
: port_(port), handler_(message::Handler::getInstance()), sender_(message::Sender::get_instance())
|
|
{
|
|
LOG_DEBUG("RpcServer 构造: 端口 {}", port_);
|
|
}
|
|
|
|
RpcServer::~RpcServer()
|
|
{
|
|
stop();
|
|
LOG_DEBUG("RpcServer 被析构，释放所有资源");
|
|
}
|
|
|
|
std::string RpcServer::build_url(int port)
|
|
{
|
|
return std::string(RpcServer::RPC_SERVER_ADDRESS) + ":" + std::to_string(port);
|
|
}
|
|
|
|
int RpcServer::start(int port)
|
|
{
|
|
if (isRunning_.load()) {
|
|
LOG_WARN("RPC 服务已在运行");
|
|
return 1;
|
|
}
|
|
|
|
port_ = port;
|
|
isRunning_ = true;
|
|
|
|
try {
|
|
cmdThread_ = std::thread(&RpcServer::run_rpc_server, this);
|
|
} catch (const std::exception &e) {
|
|
LOG_ERROR("启动 RPC 服务器失败: {}", e.what());
|
|
isRunning_ = false;
|
|
return -2;
|
|
}
|
|
#if ENABLE_WX_LOG
|
|
handler_.EnableLog();
|
|
#endif
|
|
LOG_INFO("RPC 服务器成功启动，监听端口: {}", port_);
|
|
return 0;
|
|
}
|
|
|
|
int RpcServer::stop()
|
|
{
|
|
if (!isRunning_.load()) {
|
|
LOG_WARN("RPC 服务未运行");
|
|
return 1;
|
|
}
|
|
isRunning_ = false;
|
|
|
|
handler_.UnListenPyq();
|
|
handler_.UnListenMsg();
|
|
#if ENABLE_WX_LOG
|
|
handler_.DisableLog();
|
|
#endif
|
|
nng_fini();
|
|
if (cmdThread_.joinable()) {
|
|
LOG_DEBUG("等待命令线程关闭");
|
|
cmdThread_.join();
|
|
}
|
|
LOG_DEBUG("命令线程已经关闭");
|
|
|
|
if (msgThread_.joinable()) {
|
|
LOG_DEBUG("等待消息线程关闭");
|
|
msgThread_.join();
|
|
}
|
|
LOG_DEBUG("消息线程已经关闭");
|
|
LOG_INFO("RPC 服务已停止");
|
|
return 0;
|
|
}
|
|
|
|
void RpcServer::on_message_callback()
|
|
{
|
|
try {
|
|
int rv;
|
|
nng_socket msgSock = NNG_SOCKET_INITIALIZER;
|
|
Response rsp = Response_init_default;
|
|
rsp.func = Functions_FUNC_ENABLE_RECV_TXT;
|
|
rsp.which_msg = Response_wxmsg_tag;
|
|
std::vector<uint8_t> msgBuffer(DEFAULT_BUF_SIZE);
|
|
|
|
pb_ostream_t stream = pb_ostream_from_buffer(msgBuffer.data(), msgBuffer.size());
|
|
|
|
std::string url = build_url(port_ + 1);
|
|
if ((rv = nng_pair1_open(&msgSock)) != 0) {
|
|
LOG_ERROR("nng_pair0_open error {}", nng_strerror(rv));
|
|
return;
|
|
}
|
|
|
|
if ((rv = nng_listen(msgSock, url.c_str(), NULL, 0)) != 0) {
|
|
LOG_ERROR("nng_listen error {}", nng_strerror(rv));
|
|
return;
|
|
}
|
|
|
|
if ((rv = nng_setopt_ms(msgSock, NNG_OPT_SENDTIMEO, 5000)) != 0) {
|
|
LOG_ERROR("nng_setopt_ms: {}", nng_strerror(rv));
|
|
return;
|
|
}
|
|
|
|
while (handler_.isMessageListening()) {
|
|
std::optional<WxMsg_t> msgOpt;
|
|
{
|
|
std::unique_lock<std::mutex> lock(handler_.getMutex());
|
|
bool hasMessage
|
|
= handler_.getConditionVariable().wait_for(lock, std::chrono::milliseconds(1000), [&]() {
|
|
lock.unlock();
|
|
msgOpt = handler_.popMessage();
|
|
lock.lock();
|
|
return msgOpt.has_value();
|
|
});
|
|
|
|
if (!hasMessage) {
|
|
continue;
|
|
}
|
|
}
|
|
|
|
if (!msgOpt.has_value()) {
|
|
LOG_WARN("popMessage returned empty after wait_for success.");
|
|
continue;
|
|
}
|
|
|
|
WxMsg_t wxmsg = std::move(msgOpt.value());
|
|
rsp.msg.wxmsg.id = wxmsg.id;
|
|
rsp.msg.wxmsg.is_self = wxmsg.is_self;
|
|
rsp.msg.wxmsg.is_group = wxmsg.is_group;
|
|
rsp.msg.wxmsg.type = wxmsg.type;
|
|
rsp.msg.wxmsg.ts = wxmsg.ts;
|
|
rsp.msg.wxmsg.roomid = (char *)wxmsg.roomid.c_str();
|
|
rsp.msg.wxmsg.content = (char *)wxmsg.content.c_str();
|
|
rsp.msg.wxmsg.sender = (char *)wxmsg.sender.c_str();
|
|
rsp.msg.wxmsg.sign = (char *)wxmsg.sign.c_str();
|
|
rsp.msg.wxmsg.thumb = (char *)wxmsg.thumb.c_str();
|
|
rsp.msg.wxmsg.extra = (char *)wxmsg.extra.c_str();
|
|
rsp.msg.wxmsg.xml = (char *)wxmsg.xml.c_str();
|
|
|
|
LOG_DEBUG("Push msg: {}", wxmsg.content);
|
|
pb_ostream_t stream = pb_ostream_from_buffer(msgBuffer.data(), msgBuffer.size());
|
|
if (!pb_encode(&stream, Response_fields, &rsp)) {
|
|
LOG_ERROR("Encoding failed: {}", PB_GET_ERROR(&stream));
|
|
continue;
|
|
}
|
|
|
|
rv = nng_send(msgSock, msgBuffer.data(), stream.bytes_written, 0);
|
|
if (rv != 0) {
|
|
LOG_ERROR("msgSock-nng_send: {}", nng_strerror(rv));
|
|
}
|
|
LOG_DEBUG("Send data length {}", stream.bytes_written);
|
|
}
|
|
nng_close(msgSock);
|
|
LOG_DEBUG("Leave MSG Server.");
|
|
} catch (const std::exception &e) {
|
|
LOG_ERROR("Fatal exception in on_message_callback: {}", e.what());
|
|
} catch (...) {
|
|
LOG_ERROR("Unknown fatal exception in on_message_callback.");
|
|
}
|
|
}
|
|
|
|
bool RpcServer::start_message_listener(bool pyq, uint8_t *out, size_t *len)
|
|
{
|
|
return fill_response<Functions_FUNC_ENABLE_RECV_TXT>(out, len, [&](Response &rsp) {
|
|
rsp.msg.status = handler_.ListenMsg();
|
|
if (rsp.msg.status == 0) {
|
|
if (pyq) {
|
|
handler_.ListenPyq();
|
|
}
|
|
msgThread_ = std::thread(&RpcServer::on_message_callback, this);
|
|
}
|
|
});
|
|
}
|
|
|
|
bool RpcServer::stop_message_listener(uint8_t *out, size_t *len)
|
|
{
|
|
return fill_response<Functions_FUNC_DISABLE_RECV_TXT>(out, len, [&](Response &rsp) {
|
|
rsp.msg.status = handler_.UnListenMsg();
|
|
if (rsp.msg.status == 0) {
|
|
handler_.UnListenPyq();
|
|
if (msgThread_.joinable()) {
|
|
msgThread_.join();
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
const std::unordered_map<Functions, RpcServer::FunctionHandler> RpcServer::rpcFunctionMap = {
|
|
// clang-format off
|
|
{ Functions_FUNC_IS_LOGIN, [](const Request &r, uint8_t *out, size_t *len) { return account::rpc_is_logged_in(out, len); } },
|
|
{ Functions_FUNC_GET_SELF_WXID, [](const Request &r, uint8_t *out, size_t *len) { return account::rpc_get_self_wxid(out, len); } },
|
|
{ Functions_FUNC_GET_USER_INFO, [](const Request &r, uint8_t *out, size_t *len) { return account::rpc_get_user_info(out, len); } },
|
|
{ Functions_FUNC_GET_MSG_TYPES, [](const Request &r, uint8_t *out, size_t *len) { return RpcServer::getInstance().handler_.rpc_get_msg_types(out, len); } },
|
|
{ Functions_FUNC_ENABLE_RECV_TXT, [](const Request &r, uint8_t *out, size_t *len) { return RpcServer::getInstance().start_message_listener(r.msg.flag, out, len); } },
|
|
{ Functions_FUNC_DISABLE_RECV_TXT, [](const Request &r, uint8_t *out, size_t *len) { return RpcServer::getInstance().stop_message_listener(out, len); } },
|
|
{ Functions_FUNC_GET_CONTACTS, [](const Request &r, uint8_t *out, size_t *len) { return contact::rpc_get_contacts(out, len); } },
|
|
{ Functions_FUNC_GET_DB_NAMES, [](const Request &r, uint8_t *out, size_t *len) { return db::rpc_get_db_names(out, len); } },
|
|
{ Functions_FUNC_GET_DB_TABLES, [](const Request &r, uint8_t *out, size_t *len) { return db::rpc_get_db_tables(r.msg.str, out, len); } },
|
|
{ Functions_FUNC_GET_AUDIO_MSG, [](const Request &r, uint8_t *out, size_t *len) { return misc::rpc_get_audio(r.msg.am, out, len); } },
|
|
{ Functions_FUNC_SEND_TXT, [](const Request &r, uint8_t *out, size_t *len) { return RpcServer::getInstance().sender_.rpc_send_text(r.msg.txt, out, len); } },
|
|
{ Functions_FUNC_SEND_IMG, [](const Request &r, uint8_t *out, size_t *len) { return RpcServer::getInstance().sender_.rpc_send_image(r.msg.file, out, len); } },
|
|
{ Functions_FUNC_SEND_FILE, [](const Request &r, uint8_t *out, size_t *len) { return RpcServer::getInstance().sender_.rpc_send_file(r.msg.file, out, len); } },
|
|
{ Functions_FUNC_SEND_XML, [](const Request &r, uint8_t *out, size_t *len) { return RpcServer::getInstance().sender_.rpc_send_xml(r.msg.xml, out, len); } },
|
|
{ Functions_FUNC_SEND_EMOTION, [](const Request &r, uint8_t *out, size_t *len) { return RpcServer::getInstance().sender_.rpc_send_emotion(r.msg.file, out, len); } },
|
|
{ Functions_FUNC_SEND_RICH_TXT, [](const Request &r, uint8_t *out, size_t *len) { return RpcServer::getInstance().sender_.rpc_send_rich_text(r.msg.rt, out, len); } },
|
|
{ Functions_FUNC_SEND_PAT_MSG, [](const Request &r, uint8_t *out, size_t *len) { return RpcServer::getInstance().sender_.rpc_send_pat(r.msg.pm, out, len); } },
|
|
{ Functions_FUNC_FORWARD_MSG, [](const Request &r, uint8_t *out, size_t *len) { return RpcServer::getInstance().sender_.rpc_forward(r.msg.fm, out, len); } },
|
|
{ Functions_FUNC_EXEC_DB_QUERY, [](const Request &r, uint8_t *out, size_t *len) { return db::rpc_exec_db_query(r.msg.query, out, len); } },
|
|
{ Functions_FUNC_ACCEPT_FRIEND, [](const Request &r, uint8_t *out, size_t *len) { return contact::rpc_accept_friend(r.msg.v, out, len); } },
|
|
{ Functions_FUNC_RECV_TRANSFER, [](const Request &r, uint8_t *out, size_t *len) { return misc::rpc_receive_transfer(r.msg.tf, out, len); } },
|
|
{ Functions_FUNC_REFRESH_PYQ, [](const Request &r, uint8_t *out, size_t *len) { return misc::rpc_refresh_pyq(r.msg.ui64, out, len); } },
|
|
{ Functions_FUNC_DOWNLOAD_ATTACH, [](const Request &r, uint8_t *out, size_t *len) { return misc::rpc_download_attachment(r.msg.att, out, len); } },
|
|
// { Functions_FUNC_GET_CONTACT_INFO, [](const Request &r, uint8_t *out, size_t *len) { return contact::rpc_get_contact_info(r.msg.str, out, len); } },
|
|
{ Functions_FUNC_REVOKE_MSG, [](const Request &r, uint8_t *out, size_t *len) { return misc::rpc_revoke_message(r.msg.ui64, out, len); } },
|
|
{ Functions_FUNC_REFRESH_QRCODE, [](const Request &r, uint8_t *out, size_t *len) { return misc::rpc_get_login_url(out, len); } },
|
|
{ Functions_FUNC_DECRYPT_IMAGE, [](const Request &r, uint8_t *out, size_t *len) { return misc::rpc_decrypt_image(r.msg.dec, out, len); } },
|
|
{ Functions_FUNC_EXEC_OCR, [](const Request &r, uint8_t *out, size_t *len) { return misc::rpc_get_ocr_result(r.msg.str, out, len); } },
|
|
// { Functions_FUNC_ADD_ROOM_MEMBERS, [](const Request &r, uint8_t *out, size_t *len) { return chatroom::rpc_add_chatroom_member(r.msg.m, out, len); } },
|
|
// { Functions_FUNC_DEL_ROOM_MEMBERS, [](const Request &r, uint8_t *out, size_t *len) { return chatroom::rpc_delete_chatroom_member(r.msg.m, out, len); } },
|
|
// { Functions_FUNC_INV_ROOM_MEMBERS, [](const Request &r, uint8_t *out, size_t *len) { return chatroom::rpc_invite_chatroom_member(r.msg.m, out, len); } },
|
|
// clang-format on
|
|
};
|
|
|
|
bool RpcServer::dispatcher(uint8_t *in, size_t in_len, uint8_t *out, size_t *out_len)
|
|
{
|
|
bool ret = false;
|
|
Request req = Request_init_default;
|
|
pb_istream_t stream = pb_istream_from_buffer(in, in_len);
|
|
if (!pb_decode(&stream, Request_fields, &req)) {
|
|
LOG_ERROR("Decoding failed: {}", PB_GET_ERROR(&stream));
|
|
pb_release(Request_fields, &req);
|
|
return false;
|
|
}
|
|
|
|
LOG_DEBUG("{:#04x}[{}] length: {}", (uint8_t)req.func, magic_enum::enum_name(req.func), in_len);
|
|
|
|
auto it = RpcServer::rpcFunctionMap.find(req.func);
|
|
if (it != RpcServer::rpcFunctionMap.end()) {
|
|
ret = it->second(req, out, out_len);
|
|
} else {
|
|
LOG_ERROR("[未知方法]");
|
|
}
|
|
|
|
pb_release(Request_fields, &req);
|
|
return ret;
|
|
}
|
|
|
|
void RpcServer::run_rpc_server()
|
|
{
|
|
int rv = 0;
|
|
nng_socket cmdSock = NNG_SOCKET_INITIALIZER;
|
|
std::string url = build_url(port_);
|
|
|
|
if ((rv = nng_pair1_open(&cmdSock)) != 0) {
|
|
LOG_ERROR("nng_pair1_open error: {}", nng_strerror(rv));
|
|
return;
|
|
}
|
|
|
|
if ((rv = nng_listen(cmdSock, url.c_str(), nullptr, 0)) != 0) {
|
|
LOG_ERROR("nng_listen error: {}", nng_strerror(rv));
|
|
nng_close(cmdSock);
|
|
nng_fini();
|
|
return;
|
|
}
|
|
|
|
if ((rv = nng_setopt_ms(cmdSock, NNG_OPT_SENDTIMEO, 1000)) != 0) {
|
|
LOG_ERROR("nng_setopt_ms error: {}", nng_strerror(rv));
|
|
nng_close(cmdSock);
|
|
nng_fini();
|
|
return;
|
|
}
|
|
|
|
LOG_INFO("CMD Server listening on {}", url);
|
|
std::vector<uint8_t> cmdBuffer(DEFAULT_BUF_SIZE);
|
|
|
|
while (isRunning_.load()) {
|
|
uint8_t *in = nullptr;
|
|
size_t in_len, out_len = cmdBuffer.size();
|
|
|
|
rv = nng_recv(cmdSock, &in, &in_len, NNG_FLAG_ALLOC);
|
|
if (rv != 0) {
|
|
LOG_ERROR("cmdSock-nng_recv error: {}", nng_strerror(rv));
|
|
break;
|
|
}
|
|
|
|
try {
|
|
if (dispatcher(in, in_len, cmdBuffer.data(), &out_len)) {
|
|
LOG_DEBUG("Send data length {}", out_len);
|
|
|
|
rv = nng_send(cmdSock, cmdBuffer.data(), out_len, 0);
|
|
if (rv != 0) {
|
|
LOG_ERROR("cmdSock-nng_send: {}", nng_strerror(rv));
|
|
}
|
|
} else { // 处理失败情况
|
|
LOG_ERROR("Dispatcher failed...");
|
|
rv = nng_send(cmdSock, cmdBuffer.data(), 0, 0);
|
|
if (rv != 0) {
|
|
LOG_ERROR("cmdSock-nng_send: {}", nng_strerror(rv));
|
|
}
|
|
}
|
|
} catch (const std::exception &e) {
|
|
LOG_ERROR(util::gb2312_to_utf8(e.what()));
|
|
} catch (...) {
|
|
LOG_ERROR("Unknown exception.");
|
|
}
|
|
|
|
nng_free(in, in_len);
|
|
}
|
|
nng_close(cmdSock);
|
|
LOG_DEBUG("Leave RunRpcServer");
|
|
}
|