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.h | 53 ++++++++++++++++++++++++++++++++++++++++++++--------- 1 files changed, 44 insertions(+), 9 deletions(-) diff --git a/src/topic_node.h b/src/topic_node.h index b018807..1bb3611 100644 --- a/src/topic_node.h +++ b/src/topic_node.h @@ -46,6 +46,7 @@ bool Heartbeat(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms); bool Heartbeat(const int timeout_ms); bool QueryTopicAddress(BHAddress &dest, MsgQueryTopic &query, MsgQueryTopicReply &reply_body, const int timeout_ms); + bool QueryProcs(BHAddress &dest, MsgQueryProc &query, MsgQueryProcReply &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; @@ -78,7 +79,7 @@ MQId ssn() { return SockNode().id(); } 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); + int QueryTopicServers(const Topic &topic, std::vector<NodeAddress> &addr, const int timeout_ms); const std::string &proc_id() { return info_.proc_id(); } typedef BHAddress Address; @@ -86,7 +87,11 @@ { class Impl { - typedef std::unordered_map<Topic, Address> Records; + struct TimedRec { + Address addr_; + int64_t timestamp_; + }; + typedef std::unordered_map<Topic, TimedRec> Records; Records records_; public: @@ -94,15 +99,18 @@ { auto pos = records_.find(topic); if (pos != records_.end()) { - addr = pos->second; - return true; - } else { - return false; + if (NowSec() - pos->second.timestamp_ < NodeTimeoutSec() / 2) { + addr = pos->second.addr_; + return true; + } else { + LOG_TRACE() << "topic dest cache timeout."; + } } + return false; } bool Store(const Topic &topic, const Address &addr) { - records_[topic] = addr; + records_[topic] = {addr, NowSec()}; return true; } }; @@ -139,6 +147,7 @@ } enum State { + eStateUninited, eStateUnregistered, eStateOnline, eStateOffline // heartbeat fail. @@ -146,15 +155,41 @@ 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() { return Init() && state() == eStateOnline; } + bool IsOnline() { return state() == eStateOnline; } bool Init(); - bool Valid() const { return !sockets_.empty(); } + bool Valid() const { return state() != eStateUninited; } std::mutex mutex_; MQId ssn_id_ = 0; std::atomic<State> state_; int proc_index_ = -1; TopicQueryCache topic_query_cache_; + + class RecvQ + { + public: + void Write(BHMsgHead &&head, std::string &&body) { q_.push_back({std::move(head), std::move(body)}); } + bool Read(BHMsgHead &head, std::string &body) + { + if (q_.empty()) { + return false; + } else { + head = std::move(q_.front().head); + body = std::move(q_.front().body); + q_.pop_front(); + return true; + } + } + + private: + struct MsgData { + BHMsgHead head; + std::string body; + }; + std::deque<MsgData> q_; + }; + Synced<RecvQ> server_buffer_; + Synced<RecvQ> sub_buffer_; }; #endif // end of include guard: TOPIC_NODE_YVKWA6TF -- Gitblit v1.8.0