From 056f71f24cefaf88f2a93714c6678c03ed5f1e0e Mon Sep 17 00:00:00 2001
From: zhangmeng <775834166@qq.com>
Date: 星期五, 02 七月 2021 16:54:33 +0800
Subject: [PATCH] fixed to adapt gcc-5.4 & glibc-2.25

---
 box/center_topic_node.cpp |   67 ++++++++++++++++++++++++++-------
 1 files changed, 52 insertions(+), 15 deletions(-)

diff --git a/box/center_topic_node.cpp b/box/center_topic_node.cpp
index 82b38ca..1f4103f 100644
--- a/box/center_topic_node.cpp
+++ b/box/center_topic_node.cpp
@@ -29,7 +29,8 @@
 
 namespace
 {
-const std::string &kTopicQueryProc = "@center_query_procs";
+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(); }
 
@@ -77,28 +84,58 @@
 	MsgCommonReply reply;
 
 	ProcInfo info;
-	info.set_proc_id("#center.node");
+	info.set_proc_id("@center.node");
 	info.set_name("center node");
-	if (!pnode_->Register(info, reply, timeout)) {
+	Json jinfo;
+	jinfo.put("description", "some center services. Other nodes may use topics to use them.");
+	info.set_public_info(jinfo.dump());
+	if (!pnode_->DoRegister(true, info, reply, timeout)) {
 		throw std::runtime_error("center node register failed.");
 	}
 
-	MsgTopicList topics;
-	topics.add_topic_list(kTopicQueryProc);
-	if (!pnode_->ServerRegisterRPC(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) {
 		auto reply = MakeReply<MsgRequestTopicReply>(eSuccess);
 		if (request.topic() == kTopicQueryProc) {
-			auto data = (*pscenter_)->QueryProc(request.data());
+			std::string id;
+			if (!request.data().empty()) {
+				Json json;
+				if (json.parse(request.data())) {
+					id = json.get("proc_id", "");
+				}
+			}
+			auto data = (*pscenter_)->QueryProc(id);
 			*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;
@@ -110,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