From 58d904a328c0d849769b483e901a0be9426b8209 Mon Sep 17 00:00:00 2001
From: liuxiaolong <liuxiaolong@aiotlink.com>
Date: 星期二, 20 七月 2021 20:20:44 +0800
Subject: [PATCH] 调整Request C.BHFree的位置

---
 box/center_topic_node.cpp |   49 ++++++++++++++++++++++++++++++++++++++-----------
 1 files changed, 38 insertions(+), 11 deletions(-)

diff --git a/box/center_topic_node.cpp b/box/center_topic_node.cpp
index 5c8df7a..1f4103f 100644
--- a/box/center_topic_node.cpp
+++ b/box/center_topic_node.cpp
@@ -30,6 +30,7 @@
 namespace
 {
 const std::string &kTopicQueryProc = "#center_query_procs";
+const std::string &kTopicNotifyRemoteInfo = "pub-allRegisterInfo-to-center";
 
 std::string ToJson(const MsgQueryProcReply &qpr)
 {
@@ -43,11 +44,17 @@
 		proc.put("name", info.proc().name());
 		proc.put("publicInfo", info.proc().public_info());
 		proc.put("online", info.online());
-		Json topics = Json::Array();
-		for (auto &t : info.topics().topic_list()) {
-			topics.push_back(t);
-		}
-		proc.put("topics", topics);
+		auto AddTopics = [&](auto &name, auto &topic_list) {
+			Json topics = Json::Array();
+			for (auto &t : topic_list) {
+				topics.push_back(t);
+			}
+			proc.put(name, topics);
+		};
+		AddTopics("service", info.service().topic_list());
+		AddTopics("local_sub", info.local_sub().topic_list());
+		AddTopics("net_sub", info.net_sub().topic_list());
+
 		list.push_back(proc);
 	}
 	return json.dump(0);
@@ -56,7 +63,7 @@
 } // namespace
 
 CenterTopicNode::CenterTopicNode(CenterPtr center, SharedMemory &shm) :
-    pscenter_(center), pnode_(new TopicNode(shm)), run_(false) {}
+    pscenter_(center), pnode_(new TopicNode(shm, 200)), run_(false) {}
 
 CenterTopicNode::~CenterTopicNode() { Stop(); }
 
@@ -86,10 +93,16 @@
 		throw std::runtime_error("center node register failed.");
 	}
 
-	MsgTopicList topics;
-	topics.add_topic_list(kTopicQueryProc);
-	if (!pnode_->DoServerRegisterRPC(true, topics, reply, timeout)) {
+	MsgTopicList services;
+	services.add_topic_list(kTopicQueryProc);
+	if (!pnode_->DoServerRegisterRPC(true, services, reply, timeout)) {
 		throw std::runtime_error("center node register topics failed.");
+	}
+	MsgTopicList subs;
+
+	subs.add_topic_list(kTopicNotifyRemoteInfo);
+	if (!pnode_->Subscribe(subs, reply, timeout)) {
+		throw std::runtime_error("center node subscribe topics failed.");
 	}
 
 	auto onRequest = [this](void *src_info, std::string &client_proc_id, MsgRequestTopic &request) {
@@ -106,9 +119,23 @@
 			*reply.mutable_errmsg() = data.errmsg();
 			reply.set_data(ToJson(data));
 		} else {
-			SetError(*reply.mutable_errmsg(), eInvalidInput, "not supported topic" + request.topic());
+			SetError(*reply.mutable_errmsg(), eInvalidInput, "invalid topic: " + request.topic());
 		}
 		pnode_->ServerSendReply(src_info, reply);
+	};
+
+	auto OnSubRecv = [&](const std::string &proc_id, const MsgPublish &data) {
+		if (data.topic() == kTopicNotifyRemoteInfo) {
+			// parse other data.
+			// LOG_DEBUG() << "center got net info.";
+			ssjson::Json js;
+			if (js.parse(data.data())) {
+				if (js.is_array()) {
+					auto &center = *pscenter_;
+					center->ParseNetInfo(js);
+				}
+			}
+		}
 	};
 
 	bool cur = false;
@@ -120,7 +147,7 @@
 			}
 		};
 		std::thread(heartbeat).swap(worker_);
-		return pnode_->ServerStart(onRequest);
+		return pnode_->ServerStart(onRequest) && pnode_->SubscribeStartWorker(OnSubRecv);
 	} else {
 		return false;
 	}

--
Gitblit v1.8.0