From 65a230ec6ccb61c3ce6816730da2106f07f40b4a Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期五, 23 四月 2021 18:42:41 +0800
Subject: [PATCH] add api, Unregister, QueryTopicAddress.

---
 proto/source/bhome_msg_api.proto |   21 ++++
 utest/api_test.cpp               |    2 
 api/bhsgo/bhome_node_test.go     |   18 +++
 box/center.cpp                   |   36 ++++--
 src/proto.h                      |    1 
 api/bhsgo/bhome_node.go          |   25 ++++
 proto/source/bhome_msg.proto     |    2 
 src/topic_node.h                 |    4 
 src/bh_api.h                     |   12 ++
 src/bh_api.cpp                   |    8 +
 src/topic_node.cpp               |  124 ++++++++++++++----------
 11 files changed, 184 insertions(+), 69 deletions(-)

diff --git a/api/bhsgo/bhome_node.go b/api/bhsgo/bhome_node.go
index 35d8681..f771f08 100644
--- a/api/bhsgo/bhome_node.go
+++ b/api/bhsgo/bhome_node.go
@@ -8,8 +8,9 @@
 import "C"
 
 import (
-	bh "basic.com/valib/bhshmq.git/proto/source/bhome_msg"
 	"unsafe"
+
+	bh "basic.com/valib/bhshmq.git/proto/source/bhome_msg"
 )
 
 func getPtr(n *[]byte) unsafe.Pointer {
@@ -46,8 +47,8 @@
 	return bhApiIn1Out1(C.FBHApiIn1Out1(C.BHSubscribeTopics), data, reply, timeout_ms)
 }
 
-func Heartbeat(topics *bh.ProcInfo, reply *bh.MsgCommonReply, timeout_ms int) bool {
-	data, _ := topics.Marshal()
+func Heartbeat(proc *bh.ProcInfo, reply *bh.MsgCommonReply, timeout_ms int) bool {
+	data, _ := proc.Marshal()
 	return bhApiIn1Out1(C.FBHApiIn1Out1(C.BHHeartbeat), data, reply, timeout_ms)
 }
 
@@ -55,6 +56,24 @@
 	return C.BHHeartbeatEasy(C.int(timeout_ms)) > 0
 }
 
+func Unregister(proc *bh.ProcInfo, reply *bh.MsgCommonReply, timeout_ms int) bool {
+	data, _ := proc.Marshal()
+	return bhApiIn1Out1(C.FBHApiIn1Out1(C.BHUnregister), data, reply, timeout_ms)
+}
+
+func QueryTopicAddress(topic *bh.MsgQueryTopic, reply *bh.MsgQueryTopicReply, timeout_ms int) bool {
+	data, _ := topic.Marshal()
+	creply := unsafe.Pointer(nil)
+	creply_len := C.int(0)
+	defer C.BHFree(creply, creply_len)
+	r := C.BHQueryTopicAddress(getPtr(&data), C.int(len(data)), &creply, &creply_len, C.int(timeout_ms)) > 0
+	if r {
+		reply.Unmarshal(C.GoBytes(creply, creply_len))
+	}
+	return r
+
+}
+
 func Publish(pub *bh.MsgPublish, timeout_ms int) bool {
 	data, _ := pub.Marshal()
 	return C.BHPublish(getPtr(&data), C.int(len(data)), C.int(timeout_ms)) > 0
diff --git a/api/bhsgo/bhome_node_test.go b/api/bhsgo/bhome_node_test.go
index b00b1fe..cc1966e 100644
--- a/api/bhsgo/bhome_node_test.go
+++ b/api/bhsgo/bhome_node_test.go
@@ -1,11 +1,12 @@
 package bhsgo
 
 import (
-	bh "basic.com/valib/bhshmq.git/proto/source/bhome_msg"
 	"fmt"
 	"testing"
 	"time"
 	"unsafe"
+
+	bh "basic.com/valib/bhshmq.git/proto/source/bhome_msg"
 )
 
 func ServerCallback(src unsafe.Pointer, proc_id *string, req *bh.MsgRequestTopic) {
@@ -37,6 +38,21 @@
 		t.Log("register error")
 		return
 	}
+	r = Unregister(&proc, &reply, 1000)
+	if r {
+		fmt.Println("Unregister ok")
+	} else {
+		fmt.Println("Unregister failed")
+	}
+
+	r = Register(&proc, &reply, 1000)
+	if r {
+		fmt.Println("register ok")
+	} else {
+		fmt.Println("register failed")
+		t.Log("register error")
+		return
+	}
 
 	r = HeartbeatEasy(1000)
 	if r {
diff --git a/box/center.cpp b/box/center.cpp
index 0f547e9..badfbfe 100644
--- a/box/center.cpp
+++ b/box/center.cpp
@@ -166,6 +166,18 @@
 		return HandleMsg<MsgCommonReply, Func>(head, op);
 	}
 
+	MsgCommonReply Unregister(const BHMsgHead &head, MsgUnregister &msg)
+	{
+		return HandleMsg(
+		    head, [&](Node node) -> MsgCommonReply {
+			    NodeInfo &ni = *node;
+			    auto now = NowSec(); // just set to offline.
+			    ni.state_.timestamp_ = now - offline_time_;
+			    ni.state_.UpdateState(now, offline_time_, kill_time_);
+			    return MakeReply(eSuccess);
+		    });
+	}
+
 	MsgCommonReply RegisterRPC(const BHMsgHead &head, MsgRegisterRPC &msg)
 	{
 		return HandleMsg(
@@ -206,20 +218,17 @@
 		auto query = [&](Node self) -> MsgQueryTopicReply {
 			auto pos = service_map_.find(req.topic());
 			if (pos != service_map_.end() && !pos->second.empty()) {
-				// now just find first one.
-				const TopicDest &dest = *(pos->second.begin());
-				Node dest_node(dest.weak_node_.lock());
-				if (!dest_node) {
-					service_map_.erase(pos);
-					return MakeReply<Reply>(eOffline, "topic server offline.");
-				} else if (!Valid(*dest_node)) {
-					return MakeReply<Reply>(eNoRespond, "topic server not responding.");
-				} else {
-					MsgQueryTopicReply reply = MakeReply<Reply>(eSuccess);
-					reply.mutable_address()->set_mq_id(dest.mq_);
-					return reply;
+				auto &clients = pos->second;
+				Reply reply = MakeReply<Reply>(eSuccess);
+				for (auto &dest : clients) {
+					Node dest_node(dest.weak_node_.lock());
+					if (dest_node && Valid(*dest_node)) {
+						auto node_addr = reply.add_node_address();
+						node_addr->set_proc_id(dest_node->proc_.proc_id());
+						node_addr->mutable_addr()->set_mq_id(dest.mq_);
+					}
 				}
-
+				return reply;
 			} else {
 				return MakeReply<Reply>(eNotFound, "topic server not found.");
 			}
@@ -433,6 +442,7 @@
 		switch (head.type()) {
 			CASE_ON_MSG_TYPE(Register);
 			CASE_ON_MSG_TYPE(Heartbeat);
+			CASE_ON_MSG_TYPE(Unregister);
 
 			CASE_ON_MSG_TYPE(RegisterRPC);
 			CASE_ON_MSG_TYPE(QueryTopic);
diff --git a/proto/source/bhome_msg.proto b/proto/source/bhome_msg.proto
index aabe372..6a4942d 100644
--- a/proto/source/bhome_msg.proto
+++ b/proto/source/bhome_msg.proto
@@ -44,6 +44,8 @@
 	// kMsgTypeSubscribeReply = 23;
 	kMsgTypeUnsubscribe = 24;
 	// kMsgTypeUnsubscribeReply = 25;
+	kMsgTypeUnregister = 26;
+	// kMsgTypeUnregisterReply = 27;
 
 }
 
diff --git a/proto/source/bhome_msg_api.proto b/proto/source/bhome_msg_api.proto
index fd1ae8f..838c228 100644
--- a/proto/source/bhome_msg_api.proto
+++ b/proto/source/bhome_msg_api.proto
@@ -51,6 +51,11 @@
 	repeated BHAddress addrs = 2;
 }
 
+message MsgUnregister
+{
+	ProcInfo proc = 1;
+}
+
 message MsgHeartbeat
 {
 	ProcInfo proc = 1;
@@ -62,5 +67,19 @@
 
 message MsgQueryTopicReply {
 	ErrorMsg errmsg = 1;
-	BHAddress address = 2;
+
+message BHNodeAddress {
+	bytes proc_id = 1;
+	BHAddress addr = 2;
+}
+	repeated BHNodeAddress node_address = 2;
+}
+
+message MsgQueryProc {
+	bytes proc_id = 1;
+}
+
+message MsgQueryProcReply {
+	ErrorMsg errmsg = 1;
+	repeated ProcInfo proc = 2;
 }
diff --git a/src/bh_api.cpp b/src/bh_api.cpp
index f0ba26d..5c424e0 100644
--- a/src/bh_api.cpp
+++ b/src/bh_api.cpp
@@ -104,6 +104,10 @@
 {
 	return BHApiIn1Out1<ProcInfo>(&TopicNode::Register, proc_info, proc_info_len, reply, reply_len, timeout_ms);
 }
+int BHUnregister(const void *proc_info, const int proc_info_len, void **reply, int *reply_len, const int timeout_ms)
+{
+	return BHApiIn1Out1<ProcInfo>(&TopicNode::Unregister, proc_info, proc_info_len, reply, reply_len, timeout_ms);
+}
 
 int BHHeartbeatEasy(const int timeout_ms)
 {
@@ -120,6 +124,10 @@
 	return BHApiIn1Out1<MsgTopicList>(&TopicNode::ServerRegisterRPC, topics, topics_len, reply, reply_len, timeout_ms);
 }
 
+int BHQueryTopicAddress(const void *topics, const int topics_len, void **reply, int *reply_len, const int timeout_ms)
+{
+	return BHApiIn1Out1<MsgQueryTopic, MsgQueryTopicReply>(&TopicNode::QueryTopicAddress, topics, topics_len, reply, reply_len, timeout_ms);
+}
 int BHSubscribeTopics(const void *topics, const int topics_len, void **reply, int *reply_len, const int timeout_ms)
 {
 	return BHApiIn1Out1<MsgTopicList>(&TopicNode::Subscribe, topics, topics_len, reply, reply_len, timeout_ms);
diff --git a/src/bh_api.h b/src/bh_api.h
index 33a70cb..4d6846e 100644
--- a/src/bh_api.h
+++ b/src/bh_api.h
@@ -24,12 +24,24 @@
                int *reply_len,
                const int timeout_ms);
 
+int BHUnregister(const void *proc_info,
+                 const int proc_info_len,
+                 void **reply,
+                 int *reply_len,
+                 const int timeout_ms);
+
 int BHRegisterTopics(const void *topics,
                      const int topics_len,
                      void **reply,
                      int *reply_len,
                      const int timeout_ms);
 
+int BHQueryTopicAddress(const void *topics,
+                        const int topics_len,
+                        void **reply,
+                        int *reply_len,
+                        const int timeout_ms);
+
 int BHSubscribeTopics(const void *topics,
                       const int topics_len,
                       void **reply,
diff --git a/src/proto.h b/src/proto.h
index 2557f8e..c30b4fd 100644
--- a/src/proto.h
+++ b/src/proto.h
@@ -38,6 +38,7 @@
 
 BHOME_SIMPLE_MAP_MSG(CommonReply);
 BHOME_SIMPLE_MAP_MSG(Register);
+BHOME_SIMPLE_MAP_MSG(Unregister);
 BHOME_SIMPLE_MAP_MSG(RegisterRPC);
 BHOME_SIMPLE_MAP_MSG(Heartbeat);
 BHOME_SIMPLE_MAP_MSG(QueryTopic);
diff --git a/src/topic_node.cpp b/src/topic_node.cpp
index a5d48b7..24bc4bb 100644
--- a/src/topic_node.cpp
+++ b/src/topic_node.cpp
@@ -120,6 +120,39 @@
 		return IsOnline();
 	}
 }
+bool TopicNode::Unregister(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
+{
+	info_.Clear();
+	state_cas(eStateOnline, eStateOffline);
+
+	auto &sock = SockNode();
+	MsgUnregister body;
+	body.mutable_proc()->Swap(&proc);
+
+	auto head(InitMsgHead(GetType(body), body.proc().proc_id()));
+	AddRoute(head, sock.id());
+
+	auto CheckResult = [this](MsgI &msg, BHMsgHead &head, MsgCommonReply &rbody) {
+		bool r = head.type() == kMsgTypeCommonReply &&
+		         msg.ParseBody(rbody) &&
+		         IsSuccess(rbody.errmsg().errcode());
+		return r;
+	};
+
+	if (timeout_ms == 0) {
+		auto onResult = [this, CheckResult](ShmSocket &socket, MsgI &imsg, BHMsgHead &head) {
+			MsgCommonReply body;
+			CheckResult(imsg, head, body);
+		};
+		return sock.Send(&BHTopicCenterAddress(), head, body, onResult);
+	} else {
+		MsgI reply;
+		DEFER1(reply.Release(););
+		BHMsgHead reply_head;
+		bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms);
+		return r && CheckResult(reply, reply_head, reply_body);
+	}
+}
 
 bool TopicNode::Heartbeat(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
 {
@@ -152,6 +185,25 @@
 	proc.set_proc_id(proc_id());
 	MsgCommonReply reply_body;
 	return Heartbeat(proc, reply_body, timeout_ms);
+}
+
+bool TopicNode::QueryTopicAddress(MsgQueryTopic &query, MsgQueryTopicReply &reply_body, const int timeout_ms)
+{
+	if (!IsOnline()) {
+		SetLastError(eNotRegistered, "Not Registered.");
+		return false;
+	}
+	auto &sock = SockNode();
+
+	BHMsgHead head(InitMsgHead(GetType(query), proc_id()));
+	AddRoute(head, sock.id());
+
+	MsgI reply;
+	DEFER1(reply.Release());
+	BHMsgHead reply_head;
+	return (sock.SendAndRecv(&BHTopicCenterAddress(), head, query, reply, reply_head, timeout_ms) &&
+	        reply_head.type() == kMsgTypeQueryTopicReply &&
+	        reply.ParseBody(reply_body));
 }
 
 bool TopicNode::ServerRegisterRPC(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms)
@@ -317,32 +369,7 @@
 
 	try {
 		BHAddress addr;
-#if 1
 		return (ClientQueryRPCTopic(req.topic(), addr, 3000)) && SendTo(addr, req, cb);
-#else
-		if (topic_query_cache_.Pick(req.topic(), addr)) {
-			return SendTo(addr, req, cb);
-		}
-
-		auto &sock = SockClient();
-		MsgQueryTopic query;
-		query.set_topic(req.topic());
-		BHMsgHead head(InitMsgHead(GetType(query), proc_id()));
-		AddRoute(head, sock.id());
-
-		auto onQueryResult = [this, SendTo, req, cb](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) {
-			MsgQueryTopicReply rep;
-			if (head.type() == kMsgTypeQueryTopicReply && imsg.ParseBody(rep)) {
-				auto &addr = rep.address();
-				if (!addr.mq_id().empty()) {
-					topic_query_cache_.Store(req.topic(), addr);
-					SendTo(addr, req, cb);
-				}
-			}
-		};
-		return sock.Send(&BHTopicCenterAddress(), head, query, std::move(onQueryResult));
-#endif
-
 	} catch (...) {
 		SetLastError(eError, "internal error.");
 		return false;
@@ -384,6 +411,22 @@
 	return false;
 }
 
+int TopicNode::QueryRPCTopics(const Topic &topic, std::vector<NodeAddress> &addr, const int timeout_ms)
+{
+	int n = 0;
+	MsgQueryTopic query;
+	query.set_topic(topic);
+	MsgQueryTopicReply rep;
+	if (QueryTopicAddress(query, rep, timeout_ms)) {
+		auto &ls = rep.node_address();
+		n = ls.size();
+		for (auto &na : ls) {
+			addr.push_back(na);
+		}
+	}
+	return n;
+}
+
 bool TopicNode::ClientQueryRPCTopic(const Topic &topic, BHAddress &addr, const int timeout_ms)
 {
 	if (!IsOnline()) {
@@ -391,35 +434,16 @@
 		return false;
 	}
 
-	auto &sock = SockClient();
-
 	if (topic_query_cache_.Find(topic, addr)) {
 		return true;
 	}
-
-	MsgQueryTopic query;
-	query.set_topic(topic);
-	BHMsgHead head(InitMsgHead(GetType(query), proc_id()));
-	AddRoute(head, sock.id());
-
-	MsgI reply;
-	DEFER1(reply.Release());
-	BHMsgHead reply_head;
-
-	if (sock.SendAndRecv(&BHTopicCenterAddress(), head, query, reply, reply_head, timeout_ms)) {
-		if (reply_head.type() == kMsgTypeQueryTopicReply) {
-			MsgQueryTopicReply rep;
-			if (reply.ParseBody(rep)) {
-				addr = rep.address();
-				if (addr.mq_id().empty()) {
-					return false;
-				} else {
-					topic_query_cache_.Store(topic, addr);
-					return true;
-				}
-			}
+	std::vector<NodeAddress> lst;
+	if (QueryRPCTopics(topic, lst, timeout_ms)) {
+		addr = lst.front().addr();
+		if (!addr.mq_id().empty()) {
+			topic_query_cache_.Store(topic, addr);
+			return true;
 		}
-	} else {
 	}
 	return false;
 }
diff --git a/src/topic_node.h b/src/topic_node.h
index 3371c35..20b27d2 100644
--- a/src/topic_node.h
+++ b/src/topic_node.h
@@ -41,8 +41,10 @@
 
 	// topic node
 	bool Register(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms);
+	bool Unregister(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms);
 	bool Heartbeat(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms);
 	bool Heartbeat(const int timeout_ms);
+	bool QueryTopicAddress(MsgQueryTopic &query, MsgQueryTopicReply &reply_body, const int timeout_ms);
 
 	// topic rpc server
 	typedef std::function<bool(const std::string &client_proc_id, const MsgRequestTopic &request, MsgRequestTopicReply &reply)> ServerSyncCB;
@@ -73,6 +75,8 @@
 
 private:
 	bool ClientQueryRPCTopic(const Topic &topic, BHAddress &addr, const int timeout_ms);
+	typedef MsgQueryTopicReply::BHNodeAddress NodeAddress;
+	int QueryRPCTopics(const Topic &topic, std::vector<NodeAddress> &addr, const int timeout_ms);
 	const std::string &proc_id() { return info_.proc_id(); }
 
 	typedef BHAddress Address;
diff --git a/utest/api_test.cpp b/utest/api_test.cpp
index 79236ba..5d65bd5 100644
--- a/utest/api_test.cpp
+++ b/utest/api_test.cpp
@@ -219,7 +219,7 @@
 		TLMutex mutex;
 		// CasMutex mutex;
 		auto Lock = [&]() {
-			for (int i = 0; i < 1000 * 1000 * 10; ++i) {
+			for (int i = 0; i < 10; ++i) {
 				mutex.lock();
 				mutex.unlock();
 			}

--
Gitblit v1.8.0