support tcp pub/sub.

lichao

2021-06-30 ae17d1439b35b55212c3a30712e0a60b1d6a99c0

 box/node_center.h

@@ -18,6 +18,7 @@
#ifndef NODE_CENTER_KY67RJ1Q
#define NODE_CENTER_KY67RJ1Q

#include "json.h"
#include "shm_socket.h"
#include <unordered_map>

@@ -188,7 +189,9 @@
   void OnTimer();

   // remote hosts records
   std::vector<std::string> FindRemoteSubClients(const Topic &topic);
   std::set<std::string> FindRemoteSubClients(const Topic &topic) { return net_records_.FindSubHosts(topic); }
   std::set<std::string> FindRemoteRPCServers(const Topic &topic) { return net_records_.FindRPCHosts(topic); }
   void ParseNetInfo(ssjson::Json &info) { net_records_.ParseData(info); }

private:
   void CheckNodes();
@@ -219,6 +222,33 @@
   int64_t offline_time_;
   int64_t kill_time_;
   int64_t last_check_time_;

   // net hosts info
   class NetRecords
   {
   public:
      typedef std::set<std::string> Hosts;
      void ParseData(const ssjson::Json &input);
      Hosts FindRPCHosts(const Topic &topic) { return FindHosts(topic, rpc_hosts_); }
      Hosts FindSubHosts(const Topic &topic) { return FindHosts(topic, sub_hosts_); }

   private:
      typedef std::unordered_map<Topic, Hosts> TopicMap;
      TopicMap sub_hosts_;
      TopicMap rpc_hosts_;
      Hosts FindHosts(const Topic &topic, const TopicMap &tmap)
      {
         auto pos = tmap.find(topic);
         if (pos != tmap.end()) {
            return pos->second;
         } else {
            return Hosts();
         }
      }
      std::string host_id_;
      std::string ip_;
   };
   NetRecords net_records_;
};

#endif // end of include guard: NODE_CENTER_KY67RJ1Q