From 34cd75f77d0ca94dbdba4e6cc9451fe4d33e78b3 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期三, 19 五月 2021 19:14:13 +0800
Subject: [PATCH] add api BHQueryProcs.

---
 src/topic_node.cpp |  199 ++++++++++++++++++++++++++++---------------------
 1 files changed, 112 insertions(+), 87 deletions(-)

diff --git a/src/topic_node.cpp b/src/topic_node.cpp
index 51a0ab7..8bbb929 100644
--- a/src/topic_node.cpp
+++ b/src/topic_node.cpp
@@ -70,69 +70,53 @@
 	}
 	LOG_DEBUG() << "Node Init, id " << ssn_id_;
 	auto NodeInit = [&]() {
-		auto SendInitCmd = [&]() {
-			int64_t init_cmd = ssn_id_ << 4 | EncodeCmd(eCmdNodeInit);
-			auto end_time = steady_clock::now() + 3s;
-			bool r = false;
-			do {
-				r = ShmMsgQueue::TrySend(shm(), BHTopicCenterAddress(), init_cmd);
-			} while (!r && steady_clock::now() < end_time);
-			return r;
-		};
-		if (SendInitCmd()) {
-			LOG_DEBUG() << "node send init ok";
-			auto end_time = steady_clock::now() + 3s;
-			do {
-				try {
-					//TODO recv offset, avoid query.
-					for (int i = eSockStart; i < eSockEnd; ++i) {
-						sockets_.emplace_back(new ShmSocket(shm_, false, ssn_id_ + i, kMqLen));
-					}
-					break;
-				} catch (...) {
-					sockets_.clear();
-					std::this_thread::sleep_for(100ms);
-				}
-			} while (steady_clock::now() < end_time);
+		int64_t init_request = ssn_id_ << 4 | EncodeCmd(eCmdNodeInit);
+		int64_t reply = 0;
+		if (BHNodeInit(init_request, reply) && DecodeCmd(reply) == eCmdNodeInitReply) {
+			int64_t abs_addr = reply >> 4;
+			sockets_.emplace_back(new ShmSocket(abs_addr, shm_, ssn_id_));
+			LOG_DEBUG() << "node init ok";
+		} else {
+			LOG_ERROR() << "Node Init Error";
 		}
 	};
 	if (sockets_.empty()) {
 		NodeInit();
 	}
 	if (!sockets_.empty()) {
-		auto onNodeCmd = [this](ShmSocket &socket, int64_t &val) {
-			LOG_DEBUG() << "node recv cmd: " << DecodeCmd(val);
-			switch (DecodeCmd(val)) {
-			case eCmdNodeInitReply: {
-				MsgI msg(val >> 4);
-				DEFER1(msg.Release());
+		auto onMsg = [this](ShmSocket &socket, MsgI &imsg, BHMsgHead &head) {
+			LOG_DEBUG() << "node recv type: " << head.type();
+			switch (head.type()) {
+			case kMsgTypeProcInit: {
+				// reuse msg to send proc init.
 				MsgProcInit body;
+				body.set_extra_mq_num(eSockEnd - eSockStart - 1);
 				auto head = InitMsgHead(GetType(body), info_.proc_id(), ssn_id_);
 				AddRoute(head, socket);
-				if (msg.Fill(head, body)) {
-					socket.Send(BHTopicCenterAddress(), msg);
+				if (imsg.Fill(head, body)) {
+					socket.Send(BHTopicCenterAddress(), imsg);
 				}
 			} break;
-			default:
-				break;
-			}
-			return true;
-		};
-
-		// recv msgs to avoid memory leak.
-		auto onMsg = [this](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) {
-			LOG_DEBUG() << "node recv type: " << head.type();
-			if (head.type() == kMsgTypeProcInitReply) {
+			case kMsgTypeProcInitReply: {
 				LOG_DEBUG() << "got proc init reply";
 				MsgProcInitReply reply;
-				if (imsg.ParseBody(reply)) {
+				if (imsg.ParseBody(reply) && IsSuccess(reply.errmsg().errcode())) {
+					for (auto &addr : reply.extra_mqs()) {
+						LOG_DEBUG() << "add socket " << addr.abs_addr() << ", id:" << addr.mq_id();
+						sockets_.emplace_back(new ShmSocket(addr.abs_addr(), shm(), addr.mq_id()));
+					}
 					SetProcIndex(reply.proc_index());
 					this->state_ = eStateUnregistered;
+
+					ServerStart(ServerAsyncCB(), 1);
+					SubscribeStartWorker(SubDataCB(), 1);
 				}
+			} break;
+			default: break;
 			}
 			return true;
 		};
-		SockNode().Start(1, onMsg, onNodeCmd);
+		SockNode().Start(1, onMsg);
 		return true;
 	}
 	return false;
@@ -174,10 +158,10 @@
 	}
 	auto end_time = steady_clock::now() + milliseconds(timeout_ms);
 
-	while (state_ != eStateUnregistered && steady_clock::now() < end_time) {
+	while (!Valid() && steady_clock::now() < end_time) {
 		std::this_thread::yield();
 	}
-	if (state_ != eStateUnregistered) {
+	if (!Valid()) {
 		SetLastError(eError, kErrMsgNotInit);
 		return false;
 	}
@@ -308,6 +292,25 @@
 	        reply.ParseBody(reply_body));
 }
 
+bool TopicNode::QueryProcs(BHAddress &dest, MsgQueryProc &query, MsgQueryProcReply &reply_body, const int timeout_ms)
+{
+	if (!IsOnline()) {
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
+		return false;
+	}
+	auto &sock = SockNode();
+
+	BHMsgHead head(InitMsgHead(GetType(query), proc_id(), ssn()));
+	AddRoute(head, sock);
+
+	MsgI reply;
+	DEFER1(reply.Release());
+	BHMsgHead reply_head;
+	return (sock.SendAndRecv(BHTopicCenterAddress(), head, query, reply, reply_head, timeout_ms) &&
+	        reply_head.type() == kMsgTypeQueryProcReply &&
+	        reply.ParseBody(reply_body));
+}
+
 bool TopicNode::ServerRegisterRPC(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms)
 {
 	if (!IsOnline()) {
@@ -360,26 +363,32 @@
 
 bool TopicNode::ServerStart(const ServerAsyncCB &acb, int nworker)
 {
-	auto onRecv = [this, acb](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) {
-		if (head.type() != kMsgTypeRequestTopic || head.route_size() == 0) { return; }
-		MsgRequestTopic req;
-		if (!imsg.ParseBody(req)) { return; }
+	if (acb) {
+		auto onRecv = [this, acb](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) {
+			if (head.type() != kMsgTypeRequestTopic || head.route_size() == 0) { return; }
+			MsgRequestTopic req;
+			if (!imsg.ParseBody(req)) { return; }
 
-		try {
-			SrcInfo *p = new SrcInfo;
-			if (!p) {
-				throw std::runtime_error("no memory.");
+			try {
+				SrcInfo *p = new SrcInfo;
+				if (!p) {
+					throw std::runtime_error("no memory.");
+				}
+				p->route.assign(head.route().begin(), head.route().end());
+				p->msg_id = head.msg_id();
+				acb(p, *head.mutable_proc_id(), req);
+			} catch (std::exception &e) {
+				LOG_ERROR() << "error server handle msg:" << e.what();
 			}
-			p->route.assign(head.route().begin(), head.route().end());
-			p->msg_id = head.msg_id();
-			acb(p, *head.mutable_proc_id(), req);
-		} catch (std::exception &e) {
-			LOG_ERROR() << "error server handle msg:" << e.what();
-		}
-	};
+		};
 
-	auto &sock = SockServer();
-	return acb && sock.Start(onRecv, nworker);
+		return SockServer().Start(onRecv, nworker);
+	} else {
+		auto onRequest = [this](ShmSocket &socket, MsgI &msg, BHMsgHead &head) {
+			server_buffer_->Write(std::move(head), msg.body());
+		};
+		return SockServer().Start(onRequest, nworker);
+	}
 }
 
 bool TopicNode::ServerRecvRequest(void *&src_info, std::string &proc_id, MsgRequestTopic &request, const int timeout_ms)
@@ -388,13 +397,19 @@
 		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
-
-	auto &sock = SockServer();
-
-	MsgI imsg;
 	BHMsgHead head;
-	if (sock.SyncRecv(imsg, head, timeout_ms) && head.type() == kMsgTypeRequestTopic) {
-		if (imsg.ParseBody(request)) {
+	std::string body;
+	auto end_time = steady_clock::now() + milliseconds(timeout_ms);
+	while (!server_buffer_->Read(head, body)) {
+		if (steady_clock::now() < end_time) {
+			robust::QuickSleep();
+		} else {
+			return false;
+		}
+	}
+
+	if (head.type() == kMsgTypeRequestTopic) {
+		if (request.ParseFromString(body)) {
 			head.mutable_proc_id()->swap(proc_id);
 			try {
 				SrcInfo *p = new SrcInfo;
@@ -633,20 +648,24 @@
 
 bool TopicNode::SubscribeStartWorker(const SubDataCB &tdcb, int nworker)
 {
-	auto &sock = SockSub();
-
-	auto AsyncRecvProc = [this, tdcb](ShmSocket &, MsgI &imsg, BHMsgHead &head) {
-		if (head.type() == kMsgTypePublish) {
-			MsgPublish pub;
-			if (imsg.ParseBody(pub)) {
-				tdcb(head.proc_id(), pub);
+	if (tdcb) {
+		auto AsyncRecvProc = [this, tdcb](ShmSocket &, MsgI &imsg, BHMsgHead &head) {
+			if (head.type() == kMsgTypePublish) {
+				MsgPublish pub;
+				if (imsg.ParseBody(pub)) {
+					tdcb(head.proc_id(), pub);
+				}
+			} else {
+				// ignored, or dropped
 			}
-		} else {
-			// ignored, or dropped
-		}
-	};
-
-	return tdcb && sock.Start(AsyncRecvProc, nworker);
+		};
+		return SockSub().Start(AsyncRecvProc, nworker);
+	} else {
+		auto onSub = [this](ShmSocket &socket, MsgI &msg, BHMsgHead &head) {
+			sub_buffer_->Write(std::move(head), msg.body());
+		};
+		return SockSub().Start(onSub, nworker);
+	}
 }
 
 bool TopicNode::RecvSub(std::string &proc_id, MsgPublish &pub, const int timeout_ms)
@@ -656,13 +675,19 @@
 		return false;
 	}
 
-	auto &sock = SockSub();
-	MsgI msg;
-	DEFER1(msg.Release(););
 	BHMsgHead head;
+	std::string body;
+	auto end_time = steady_clock::now() + milliseconds(timeout_ms);
+	while (!sub_buffer_->Read(head, body)) {
+		if (steady_clock::now() < end_time) {
+			robust::QuickSleep();
+		} else {
+			return false;
+		}
+	}
 	//TODO error msg.
-	if (sock.SyncRecv(msg, head, timeout_ms) && head.type() == kMsgTypePublish) {
-		if (msg.ParseBody(pub)) {
+	if (head.type() == kMsgTypePublish) {
+		if (pub.ParseFromString(body)) {
 			head.mutable_proc_id()->swap(proc_id);
 			return true;
 		}

--
Gitblit v1.8.0