/* * ===================================================================================== * * Filename: topic_node.h * * Description: * * Version: 1.0 * Created: 2021年04月07日 09时05分26秒 * Revision: none * Compiler: gcc * * Author: Li Chao (), lichao@aiotlink.com * Organization: * * ===================================================================================== */ #ifndef TOPIC_NODE_YVKWA6TF #define TOPIC_NODE_YVKWA6TF #include "msg.h" #include "socket.h" #include #include #include using namespace bhome_shm; using namespace bhome_msg; // a node is a client. class TopicNode { SharedMemory &shm_; ProcInfo info_; SharedMemory &shm() { return shm_; } public: TopicNode(SharedMemory &shm); ~TopicNode(); // topic node bool Register(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); // topic rpc server typedef std::function ServerSyncCB; typedef std::function 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); // topic client typedef std::function RequestResultCB; bool ClientStartWorker(RequestResultCB const &cb, const int nworker = 2); bool ClientAsyncRequest(const MsgRequestTopic &request, std::string &msg_id, const RequestResultCB &rrcb = RequestResultCB()); bool ClientSyncRequest(const MsgRequestTopic &request, std::string &proc_id, MsgRequestTopicReply &reply, const int timeout_ms); // publish bool Publish(const MsgPublish &pub, const int timeout_ms); // subscribe typedef std::function SubDataCB; bool SubscribeStartWorker(const SubDataCB &tdcb, int nworker = 2); 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(ServerAsyncCB const &server_cb, SubDataCB const &sub_cb, RequestResultCB &client_cb, int nworker = 2); void Stop(); private: bool ClientQueryRPCTopic(const Topic &topic, BHAddress &addr, const int timeout_ms); const std::string &proc_id() { return info_.proc_id(); } typedef BHAddress Address; class TopicQueryCache { class Impl { typedef std::unordered_map Records; Records records_; public: bool Find(const Topic &topic, Address &addr) { auto pos = records_.find(topic); if (pos != records_.end()) { addr = pos->second; return true; } else { return false; } } bool Store(const Topic &topic, const Address &addr) { records_[topic] = addr; return true; } }; Synced impl_; public: bool Find(const Topic &topic, Address &addr) { return impl_->Find(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> sockets_; 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]; } 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_; TopicQueryCache topic_query_cache_; }; #endif // end of include guard: TOPIC_NODE_YVKWA6TF