From 628c1c21ffb19d8c96ed9ce89531595f9870ab1a Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期五, 23 四月 2021 18:41:02 +0800
Subject: [PATCH] add msg tag; recv all msgs before remove mq.
---
src/topic_node.cpp | 20 ++++++++++----------
1 files changed, 10 insertions(+), 10 deletions(-)
diff --git a/src/topic_node.cpp b/src/topic_node.cpp
index 8bf9cf8..a5d48b7 100644
--- a/src/topic_node.cpp
+++ b/src/topic_node.cpp
@@ -111,7 +111,7 @@
return sock.Send(&BHTopicCenterAddress(), head, body, onResult);
} else {
MsgI reply;
- DEFER1(reply.Release(shm_););
+ DEFER1(reply.Release(););
BHMsgHead reply_head;
bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms);
if (r) {
@@ -139,7 +139,7 @@
return sock.Send(&BHTopicCenterAddress(), head, body);
} else {
MsgI reply;
- DEFER1(reply.Release(shm_););
+ DEFER1(reply.Release(););
BHMsgHead reply_head;
bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms);
r = r && reply_head.type() == kMsgTypeCommonReply && reply.ParseBody(reply_body);
@@ -172,7 +172,7 @@
return sock.Send(&BHTopicCenterAddress(), head, body);
} else {
MsgI reply;
- DEFER1(reply.Release(shm_););
+ DEFER1(reply.Release(););
BHMsgHead reply_head;
bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms);
r = r && reply_head.type() == kMsgTypeCommonReply;
@@ -283,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.");
@@ -349,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.");
@@ -366,7 +366,7 @@
head.set_topic(request.topic());
MsgI reply_msg;
- DEFER1(reply_msg.Release(shm_););
+ DEFER1(reply_msg.Release(););
BHMsgHead reply_head;
if (sock.SendAndRecv(addr.mq_id().data(), head, request, reply_msg, reply_head, timeout_ms) &&
@@ -403,7 +403,7 @@
AddRoute(head, sock.id());
MsgI reply;
- DEFER1(reply.Release(shm_));
+ DEFER1(reply.Release());
BHMsgHead reply_head;
if (sock.SendAndRecv(&BHTopicCenterAddress(), head, query, reply, reply_head, timeout_ms)) {
@@ -442,7 +442,7 @@
return sock.Send(&BHTopicBusAddress(), head, pub);
} else {
MsgI reply;
- DEFER1(reply.Release(shm()););
+ DEFER1(reply.Release(););
BHMsgHead reply_head;
MsgCommonReply reply_body;
return sock.SendAndRecv(&BHTopicBusAddress(), head, pub, reply, reply_head, timeout_ms) &&
@@ -475,7 +475,7 @@
return sock.Send(&BHTopicBusAddress(), head, sub);
} else {
MsgI reply;
- DEFER1(reply.Release(shm()););
+ DEFER1(reply.Release(););
BHMsgHead reply_head;
return sock.SendAndRecv(&BHTopicBusAddress(), head, sub, reply, reply_head, timeout_ms) &&
reply_head.type() == kMsgTypeCommonReply &&
@@ -515,7 +515,7 @@
auto &sock = SockSub();
MsgI msg;
- DEFER1(msg.Release(shm()););
+ DEFER1(msg.Release(););
BHMsgHead head;
//TODO error msg.
if (sock.SyncRecv(msg, head, timeout_ms) && head.type() == kMsgTypePublish) {
--
Gitblit v1.8.0