From ad4f3dcedab29a690c5eedbb08ba1b393917db0b Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期三, 21 四月 2021 17:39:34 +0800
Subject: [PATCH] update go api.

---
 src/topic_node.h |   70 ++++++++++++++++++++---------------
 1 files changed, 40 insertions(+), 30 deletions(-)

diff --git a/src/topic_node.h b/src/topic_node.h
index 5a3b86e..b2fae5b 100644
--- a/src/topic_node.h
+++ b/src/topic_node.h
@@ -20,7 +20,9 @@
 
 #include "msg.h"
 #include "socket.h"
+#include <atomic>
 #include <memory>
+#include <vector>
 
 using namespace bhome_shm;
 using namespace bhome_msg;
@@ -34,7 +36,6 @@
 	SharedMemory &shm() { return shm_; }
 
 public:
-	typedef std::function<void(std::string &proc_id, const void *data, const int len)> DataCB;
 	TopicNode(SharedMemory &shm);
 	~TopicNode();
 
@@ -44,8 +45,10 @@
 	bool Heartbeat(const int timeout_ms);
 
 	// topic rpc server
-	typedef std::function<bool(const std::string &client_proc_id, const MsgRequestTopic &request, MsgRequestTopicReply &reply)> ServerCB;
-	bool ServerStart(ServerCB const &cb, const int nworker = 2);
+	typedef std::function<bool(const std::string &client_proc_id, const MsgRequestTopic &request, MsgRequestTopicReply &reply)> ServerSyncCB;
+	typedef std::function<void(void *src_info, std::string &client_proc_id, MsgRequestTopic &request)> ServerAsyncCB;
+	bool ServerStart(ServerSyncCB const &cb, const int nworker = 2);
+	bool ServerStart(ServerAsyncCB const &cb, const int nworker = 2);
 	bool ServerRegisterRPC(MsgTopicList &topics, MsgCommonReply &reply, const int timeout_ms);
 	bool ServerRecvRequest(void *&src_info, std::string &proc_id, MsgRequestTopic &request, const int timeout_ms);
 	bool ServerSendReply(void *src_info, const MsgRequestTopicReply &reply);
@@ -65,65 +68,72 @@
 	bool Subscribe(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms);
 	bool RecvSub(std::string &proc_id, MsgPublish &pub, const int timeout_ms);
 
-	void Start(ServerCB const &server_cb, SubDataCB const &sub_cb, RequestResultCB &client_cb, int nworker = 2);
+	void Start(ServerAsyncCB const &server_cb, SubDataCB const &sub_cb, RequestResultCB &client_cb, int nworker = 2);
 	void Stop();
 
 private:
-	bool ClientQueryRPCTopic(const Topic &topic, bhome::msg::BHAddress &addr, const int timeout_ms);
+	bool ClientQueryRPCTopic(const Topic &topic, BHAddress &addr, const int timeout_ms);
 	const std::string &proc_id() { return info_.proc_id(); }
 
-	typedef bhome_msg::BHAddress Address;
+	typedef BHAddress Address;
 	class TopicQueryCache
 	{
 		class Impl
 		{
-			typedef std::unordered_map<Topic, Address> Store;
-			Store store_;
+			typedef std::unordered_map<Topic, Address> Records;
+			Records records_;
 
 		public:
 			bool Find(const Topic &topic, Address &addr)
 			{
-				auto pos = store_.find(topic);
-				if (pos != store_.end()) {
+				auto pos = records_.find(topic);
+				if (pos != records_.end()) {
 					addr = pos->second;
 					return true;
 				} else {
 					return false;
 				}
 			}
-			bool Update(const Topic &topic, const Address &addr)
+			bool Store(const Topic &topic, const Address &addr)
 			{
-				store_[topic] = addr;
+				records_[topic] = addr;
 				return true;
 			}
 		};
 		Synced<Impl> impl_;
-		// Impl &impl()
-		// {
-		// 	thread_local Impl impl;
-		// 	return impl;
-		// }
 
 	public:
 		bool Find(const Topic &topic, Address &addr) { return impl_->Find(topic, addr); }
-		bool Update(const Topic &topic, const Address &addr) { return impl_->Update(topic, addr); }
+		bool Store(const Topic &topic, const Address &addr) { return impl_->Store(topic, addr); }
 	};
 
 	// some sockets may be the same one, using functions make it easy to change.
+	enum { eSockStart,
+		   eSockNode = eSockStart,
+		   eSockPub = eSockNode,
+		   eSockServer,
+		   eSockClient,
+		   eSockSub,
+		   eSockEnd,
+	};
+	std::vector<std::unique_ptr<ShmSocket>> sockets_;
 
-	auto &SockNode() { return sock_node_; }
-	auto &SockPub() { return SockNode(); }
-	auto &SockSub() { return sock_sub_; }
-	auto &SockClient() { return sock_client_; }
-	auto &SockServer() { return sock_server_; }
-	bool IsRegistered() const { return registered_.load(); }
+	ShmSocket &SockNode() { return *sockets_[eSockNode]; }
+	ShmSocket &SockPub() { return *sockets_[eSockPub]; }
+	ShmSocket &SockSub() { return *sockets_[eSockSub]; }
+	ShmSocket &SockClient() { return *sockets_[eSockClient]; }
+	ShmSocket &SockServer() { return *sockets_[eSockServer]; }
 
-	ShmSocket sock_node_;
-	ShmSocket sock_client_;
-	ShmSocket sock_server_;
-	ShmSocket sock_sub_;
-	std::atomic<bool> registered_;
-	std::atomic<bool> registered_ever_;
+	enum State {
+		eStateUnregistered,
+		eStateOnline,
+		eStateOffline // heartbeat fail.
+	};
+	void state(const State st) { state_.store(st); }
+	void state_cas(State expected, const State val) { state_.compare_exchange_strong(expected, val); }
+	State state() const { return state_.load(); }
+	bool IsOnline() const { return state() == eStateOnline; }
+	std::atomic<State> state_;
 
 	TopicQueryCache topic_query_cache_;
 };

--
Gitblit v1.8.0