From 02ba913dc7bb5d711471b27f2ea23a897d0f2e28 Mon Sep 17 00:00:00 2001 From: lichao <lichao@aiotlink.com> Date: 星期五, 23 四月 2021 15:34:26 +0800 Subject: [PATCH] bind msgi to shm, change offset_ptr to abs offset. --- src/bh_api.cpp | 130 ++++++++++++++++++++----------------------- 1 files changed, 61 insertions(+), 69 deletions(-) diff --git a/src/bh_api.cpp b/src/bh_api.cpp index 2abe66d..f0ba26d 100644 --- a/src/bh_api.cpp +++ b/src/bh_api.cpp @@ -10,6 +10,7 @@ { TopicNode &ProcNode() { + static bool init_bind_msg_shm = MsgI::BindShm(BHomeShm()); static TopicNode node(BHomeShm()); return node; } @@ -83,44 +84,50 @@ return false; } MsgOut msg_reply; - if ((ProcNode().*mfunc)(input, msg_reply, timeout_ms)) { - return PackOutput(msg_reply, reply, reply_len); - - } else { - return false; - } + return (ProcNode().*mfunc)(input, msg_reply, timeout_ms) && + PackOutput(msg_reply, reply, reply_len); } } // namespace -bool BHRegister(const void *proc_info, const int proc_info_len, void **reply, int *reply_len, const int timeout_ms) +int BHApiIn1Out1Proxy(FBHApiIn1Out1 func, + const void *request, + const int request_len, + void **reply, + int *reply_len, + const int timeout_ms) +{ + return (*func)(request, request_len, reply, reply_len, timeout_ms); +} + +int BHRegister(const void *proc_info, const int proc_info_len, void **reply, int *reply_len, const int timeout_ms) { return BHApiIn1Out1<ProcInfo>(&TopicNode::Register, proc_info, proc_info_len, reply, reply_len, timeout_ms); } -bool BHHeartBeatEasy(const int timeout_ms) +int BHHeartbeatEasy(const int timeout_ms) { return ProcNode().Heartbeat(timeout_ms); } -bool BHHeartBeat(const void *proc_info, const int proc_info_len, void **reply, int *reply_len, const int timeout_ms) +int BHHeartbeat(const void *proc_info, const int proc_info_len, void **reply, int *reply_len, const int timeout_ms) { return BHApiIn1Out1<ProcInfo>(&TopicNode::Heartbeat, proc_info, proc_info_len, reply, reply_len, timeout_ms); } -bool BHRegisterTopics(const void *topics, const int topics_len, void **reply, int *reply_len, const int timeout_ms) +int BHRegisterTopics(const void *topics, const int topics_len, void **reply, int *reply_len, const int timeout_ms) { return BHApiIn1Out1<MsgTopicList>(&TopicNode::ServerRegisterRPC, topics, topics_len, reply, reply_len, timeout_ms); } -bool BHSubscribeTopics(const void *topics, const int topics_len, void **reply, int *reply_len, const int timeout_ms) +int BHSubscribeTopics(const void *topics, const int topics_len, void **reply, int *reply_len, const int timeout_ms) { return BHApiIn1Out1<MsgTopicList>(&TopicNode::Subscribe, topics, topics_len, reply, reply_len, timeout_ms); } -bool BHPublish(const void *msgpub, - const int msgpub_len, - const int timeout_ms) +int BHPublish(const void *msgpub, + const int msgpub_len, + const int timeout_ms) { MsgPublish pub; if (!pub.ParseFromArray(msgpub, msgpub_len)) { @@ -130,11 +137,11 @@ return ProcNode().Publish(pub, timeout_ms); } -bool BHReadSub(void **proc_id, - int *proc_id_len, - void **msgpub, - int *msgpub_len, - const int timeout_ms) +int BHReadSub(void **proc_id, + int *proc_id_len, + void **msgpub, + int *msgpub_len, + const int timeout_ms) { std::string proc; MsgPublish pub; @@ -151,19 +158,23 @@ return false; } -bool BHAsyncRequest(const void *request, - const int request_len, - void **msg_id, - int *msg_id_len) +int BHAsyncRequest(const void *remote, + const int remote_len, + const void *request, + const int request_len, + void **msg_id, + int *msg_id_len) { + BHAddress dest; MsgRequestTopic req; - if (!req.ParseFromArray(request, request_len)) { + if (!dest.ParseFromArray(remote, remote_len) || + !req.ParseFromArray(request, request_len)) { SetLastError(eInvalidInput, "invalid input."); return false; } std::string str_msg_id; MsgRequestTopicReply out_msg; - if (ProcNode().ClientAsyncRequest(req, str_msg_id)) { + if (ProcNode().ClientAsyncRequest(dest, req, str_msg_id)) { if (!msg_id || !msg_id_len) { return true; } @@ -178,22 +189,26 @@ return false; } -bool BHRequest(const void *request, - const int request_len, - void **proc_id, - int *proc_id_len, - void **reply, - int *reply_len, - const int timeout_ms) +int BHRequest(const void *remote, + const int remote_len, + const void *request, + const int request_len, + void **proc_id, + int *proc_id_len, + void **reply, + int *reply_len, + const int timeout_ms) { + BHAddress dest; MsgRequestTopic req; - if (!req.ParseFromArray(request, request_len)) { + if (!dest.ParseFromArray(remote, remote_len) || + !req.ParseFromArray(request, request_len)) { SetLastError(eInvalidInput, "invalid input."); return false; } std::string proc; MsgRequestTopicReply out_msg; - if (ProcNode().ClientSyncRequest(req, proc, out_msg, timeout_ms)) { + if (ProcNode().ClientSyncRequest(dest, req, proc, out_msg, timeout_ms)) { TmpPtr pproc(proc); if (pproc && PackOutput(out_msg, reply, reply_len)) { pproc.ReleaseTo(proc_id, proc_id_len); @@ -205,12 +220,12 @@ return false; } -bool BHReadRequest(void **proc_id, - int *proc_id_len, - void **request, - int *request_len, - void **src, - const int timeout_ms) +int BHReadRequest(void **proc_id, + int *proc_id_len, + void **request, + int *request_len, + void **src, + const int timeout_ms) { void *src_info = 0; std::string proc; @@ -228,9 +243,9 @@ return false; } -bool BHSendReply(void *src, - const void *reply, - const int reply_len) +int BHSendReply(void *src, + const void *reply, + const int reply_len) { MsgRequestTopicReply rep; if (!rep.ParseFromArray(reply, reply_len)) { @@ -240,31 +255,15 @@ return ProcNode().ServerSendReply(src, rep); } -int BHCleanUp() -{ - return 0; -} - -namespace -{ -typedef std::function<bool(const void *, const int)> ServerSender; -} // namespace - void BHStartWorker(FServerCallback server_cb, FSubDataCallback sub_cb, FClientCallback client_cb) { - TopicNode::ServerCB on_req; + TopicNode::ServerAsyncCB on_req; TopicNode::SubDataCB on_sub; TopicNode::RequestResultCB on_reply; if (server_cb) { - on_req = [server_cb](const std::string &proc_id, const MsgRequestTopic &request, MsgRequestTopicReply &reply) { + on_req = [server_cb](void *src, std::string &proc_id, const MsgRequestTopic &request) { std::string sreq(request.SerializeAsString()); - bool r = false; - ServerSender sender = [&](const void *p, const int len) { - r = reply.ParseFromArray(p, len); - return r; - }; - server_cb(proc_id.data(), proc_id.size(), sreq.data(), sreq.size(), (BHServerCallbackTag *) (&sender)); - return r; + server_cb(proc_id.data(), proc_id.size(), sreq.data(), sreq.size(), src); }; } if (sub_cb) { @@ -283,13 +282,6 @@ } ProcNode().Start(on_req, on_sub, on_reply); -} -bool BHServerCallbackReply(const BHServerCallbackTag *tag, - const void *data, - const int data_len) -{ - auto &sender = *(const ServerSender *) (tag); - return sender(data, data_len); } void BHFree(void *data, int size) -- Gitblit v1.8.0