add center topic node.

lichao

2021-05-20 0e31f38fc37216e1376d8101d1bcf7a3779279dc

 box/node_center.cpp

@@ -350,44 +350,48 @@
      return MakeReply(eSuccess);
   });
}
MsgQueryProcReply NodeCenter::QueryProc(const BHMsgHead &head, const MsgQueryProc &req)

MsgQueryProcReply NodeCenter::QueryProc(const std::string &proc_id)
{
   typedef MsgQueryProcReply Reply;
   auto query = [&](Node self) -> Reply {
      auto Add1 = [](Reply &reply, Node node) {
         auto info = reply.add_proc_list();
         *info->mutable_proc() = node->proc_;
         info->set_online(node->state_.flag_ == kStateNormal);
         for (auto &addr_topics : node->services_) {
            for (auto &topic : addr_topics.second) {
               info->mutable_topics()->add_topic_list(topic);
            }
   auto Add1 = [](Reply &reply, Node node) {
      auto info = reply.add_proc_list();
      *info->mutable_proc() = node->proc_;
      info->mutable_proc()->clear_private_info();
      info->set_online(node->state_.flag_ == kStateNormal);
      for (auto &addr_topics : node->services_) {
         for (auto &topic : addr_topics.second) {
            info->mutable_topics()->add_topic_list(topic);
         }
      };

      if (!req.proc_id().empty()) {
         auto pos = online_node_addr_map_.find(req.proc_id());
         if (pos == online_node_addr_map_.end()) {
            return MakeReply<Reply>(eNotFound, "proc not found.");
         } else {
            auto node_pos = nodes_.find(pos->second);
            if (node_pos == nodes_.end()) {
               return MakeReply<Reply>(eNotFound, "proc node not found.");
            } else {
               auto reply = MakeReply<Reply>(eSuccess);
               Add1(reply, node_pos->second);
               return reply;
            }
         }
      } else {
         Reply reply(MakeReply<Reply>(eSuccess));
         for (auto &kv : nodes_) {
            Add1(reply, kv.second);
         }
         return reply;
      }
   };

   if (!proc_id.empty()) {
      auto pos = online_node_addr_map_.find(proc_id);
      if (pos == online_node_addr_map_.end()) {
         return MakeReply<Reply>(eNotFound, "proc not found.");
      } else {
         auto node_pos = nodes_.find(pos->second);
         if (node_pos == nodes_.end()) {
            return MakeReply<Reply>(eNotFound, "proc node not found.");
         } else {
            auto reply = MakeReply<Reply>(eSuccess);
            Add1(reply, node_pos->second);
            return reply;
         }
      }
   } else {
      Reply reply(MakeReply<Reply>(eSuccess));
      for (auto &kv : nodes_) {
         Add1(reply, kv.second);
      }
      return reply;
   }
}
MsgQueryProcReply NodeCenter::QueryProc(const BHMsgHead &head, const MsgQueryProc &req)
{
   typedef MsgQueryProcReply Reply;
   auto query = [&](Node self) -> Reply { return this->QueryProc(req.proc_id()); };
   return HandleMsg<Reply>(head, query);
}