change mqid from uuid to uint64.

lichao

2021-04-25 1fbfef2a51db4a3bac9d8a5b87af94a40a913b7a

 box/center.cpp

@@ -37,9 +37,9 @@
{
public:
   typedef std::string ProcId;
   typedef std::string Address;
   typedef MQId Address;
   typedef bhome_msg::ProcInfo ProcInfo;
   typedef std::function<void(Address const &)> Cleaner;
   typedef std::function<void(Address const)> Cleaner;

private:
   enum {
@@ -84,7 +84,7 @@
      WeakNode weak_node_;
      bool operator<(const TopicDest &a) const { return mq_ < a.mq_; }
   };
   inline const std::string &SrcAddr(const BHMsgHead &head) { return head.route(0).mq_id(); }
   inline MQId SrcAddr(const BHMsgHead &head) { return head.route(0).mq_id(); }
   inline bool MatchAddr(std::set<Address> const &addrs, const Address &addr) { return addrs.find(addr) != addrs.end(); }

   NodeCenter(const std::string &id, const Cleaner &cleaner, const int64_t offline_time, const int64_t kill_time) :
@@ -182,7 +182,7 @@
   {
      return HandleMsg(
          head, [&](Node node) -> MsgCommonReply {
             auto &src = SrcAddr(head);
             auto src = SrcAddr(head);
             auto &topics = msg.topics().topic_list();
             node->services_[src].insert(topics.begin(), topics.end());
             TopicDest dest = {src, node};
@@ -240,7 +240,7 @@
   MsgCommonReply Subscribe(const BHMsgHead &head, const MsgSubscribe &msg)
   {
      return HandleMsg(head, [&](Node node) {
         auto &src = SrcAddr(head);
         auto src = SrcAddr(head);
         auto &topics = msg.topics().topic_list();
         node->subscriptions_[src].insert(topics.begin(), topics.end());
         TopicDest dest = {src, node};
@@ -253,7 +253,7 @@
   MsgCommonReply Unsubscribe(const BHMsgHead &head, const MsgUnsubscribe &msg)
   {
      return HandleMsg(head, [&](Node node) {
         auto &src = SrcAddr(head);
         auto src = SrcAddr(head);
         auto pos = node->subscriptions_.find(src);

         auto RemoveSubTopicDestRecord = [this](const Topic &topic, const TopicDest &dest) {
@@ -426,8 +426,8 @@
   auto MakeReplyer = [](ShmSocket &socket, BHMsgHead &head, const std::string &proc_id) {
      return [&](auto &&rep_body) {
         auto reply_head(InitMsgHead(GetType(rep_body), proc_id, head.msg_id()));
         auto &remote = head.route(0).mq_id();
         socket.Send(remote.data(), reply_head, rep_body);
         auto remote = head.route(0).mq_id();
         socket.Send(remote, reply_head, rep_body);
      };
   };

@@ -473,7 +473,7 @@
               if (node) {
                  // should also make sure that mq is not killed before msg expires.
                  // it would be ok if (kill_time - offline_time) is longer than expire time.
                  socket.Send(cli.mq_.data(), msg);
                  socket.Send(cli.mq_, msg);
                  ++it;
               } else {
                  it = clients.erase(it);
@@ -505,28 +505,24 @@
   return rec;
}

bool BHCenter::Install(const std::string &name, MsgHandler handler, IdleHandler idle, const std::string &mqid, const int mq_len)
bool BHCenter::Install(const std::string &name, MsgHandler handler, IdleHandler idle, const MQId mqid, const int mq_len)
{
   Centers()[name] = CenterInfo{name, handler, idle, mqid, mq_len};
   return true;
}
bool BHCenter::Install(const std::string &name, MsgHandler handler, IdleHandler idle, const MQId &mqid, const int mq_len)
{
   return Install(name, handler, idle, std::string((const char *) &mqid, sizeof(mqid)), mq_len);
}

BHCenter::BHCenter(Socket::Shm &shm)
{
   auto gc = [&](const std::string &id) {
      auto r = ShmSocket::Remove(shm, *(MQId *) id.data());
      printf("remove mq : %s\n", r ? "ok" : "failed");
   auto gc = [&](const MQId id) {
      auto r = ShmSocket::Remove(shm, id);
      printf("remove mq %ld : %s\n", id, (r ? "ok" : "failed"));
   };

   AddCenter("#bhome_center", gc);

   for (auto &kv : Centers()) {
      auto &info = kv.second;
      sockets_[info.name_] = std::make_shared<ShmSocket>(shm, *(MQId *) info.mqid_.data(), info.mq_len_);
      sockets_[info.name_] = std::make_shared<ShmSocket>(shm, info.mqid_, info.mq_len_);
   }
}