From dc86ace85e437ecb8a2e728e4dce36d02bbb8a6e Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期五, 23 四月 2021 12:59:50 +0800
Subject: [PATCH] move ref count into msg meta, only 1 poinetr now.
---
src/topic_node.cpp | 27 ++++++++++++---------------
1 files changed, 12 insertions(+), 15 deletions(-)
diff --git a/src/topic_node.cpp b/src/topic_node.cpp
index 70247dd..4f0c96f 100644
--- a/src/topic_node.cpp
+++ b/src/topic_node.cpp
@@ -37,14 +37,16 @@
} // namespace
TopicNode::TopicNode(SharedMemory &shm) :
- shm_(shm), sock_node_(shm), sock_client_(shm, kMqLen), sock_server_(shm, kMqLen), sock_sub_(shm, kMqLen), state_(eStateUnregistered)
+ shm_(shm), sockets_(eSockEnd), state_(eStateUnregistered)
{
+ for (int i = eSockStart; i < eSockEnd; ++i) {
+ sockets_[i].reset(new ShmSocket(shm_, kMqLen));
+ }
// recv msgs to avoid memory leak.
auto default_ignore_msg = [](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) { return true; };
- SockNode().Start(default_ignore_msg);
- SockClient().Start(default_ignore_msg);
- SockServer().Start(default_ignore_msg);
- SockSub().Start(default_ignore_msg);
+ for (auto &p : sockets_) {
+ p->Start(default_ignore_msg);
+ }
}
TopicNode::~TopicNode()
@@ -52,10 +54,7 @@
Stop();
SockNode().Stop();
if (state() == eStateUnregistered) {
- SockNode().Remove();
- SockClient().Remove();
- SockServer().Remove();
- SockSub().Remove();
+ for (auto &p : sockets_) { p->Remove(); }
}
}
@@ -66,16 +65,14 @@
} else if (nworker > 16) {
nworker = 16;
}
-
+ SockNode().Start();
ServerStart(server_cb, nworker);
SubscribeStartWorker(sub_cb, nworker);
ClientStartWorker(client_cb, nworker);
}
void TopicNode::Stop()
{
- SockSub().Stop();
- SockServer().Stop();
- SockClient().Stop();
+ for (auto &p : sockets_) { p->Stop(); }
}
bool TopicNode::Register(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
@@ -286,7 +283,7 @@
return SockClient().Start(onData, nworker);
}
-bool TopicNode::ClientAsyncRequest(const MsgRequestTopic &req, std::string &out_msg_id, const RequestResultCB &cb)
+bool TopicNode::ClientAsyncRequest(const BHAddress &remote_addr, const MsgRequestTopic &req, std::string &out_msg_id, const RequestResultCB &cb)
{
if (!IsOnline()) {
SetLastError(eNotRegistered, "Not Registered.");
@@ -352,7 +349,7 @@
}
}
-bool TopicNode::ClientSyncRequest(const MsgRequestTopic &request, std::string &out_proc_id, MsgRequestTopicReply &out_reply, const int timeout_ms)
+bool TopicNode::ClientSyncRequest(const BHAddress &remote_addr, const MsgRequestTopic &request, std::string &out_proc_id, MsgRequestTopicReply &out_reply, const int timeout_ms)
{
if (!IsOnline()) {
SetLastError(eNotRegistered, "Not Registered.");
--
Gitblit v1.8.0