| | |
| | | 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; |
| | |
| | | { |
| | | 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: |
| | |
| | | { |
| | | 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; |
| | | } |
| | | }; |
| | |
| | | } |
| | | |
| | | enum State { |
| | | eStateUninited, |
| | | 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() { 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_; |