add msg tag; recv all msgs before remove mq.

lichao

2021-04-23 628c1c21ffb19d8c96ed9ce89531595f9870ab1a

 src/topic_node.cpp

@@ -37,14 +37,16 @@
} // namespace

TopicNode::TopicNode(SharedMemory &shm) :
    shm_(shm), sock_node_(shm), sock_client_(shm, kMqLen), sock_server_(shm, kMqLen), sock_sub_(shm, kMqLen), state_(eStateUnregistered)
    shm_(shm), sockets_(eSockEnd), state_(eStateUnregistered)
{
   for (int i = eSockStart; i < eSockEnd; ++i) {
      sockets_[i].reset(new ShmSocket(shm_, kMqLen));
   }
   // recv msgs to avoid memory leak.
   auto default_ignore_msg = [](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) { return true; };
   SockNode().Start(default_ignore_msg);
   SockClient().Start(default_ignore_msg);
   SockServer().Start(default_ignore_msg);
   SockSub().Start(default_ignore_msg);
   for (auto &p : sockets_) {
      p->Start(default_ignore_msg);
   }
}

TopicNode::~TopicNode()
@@ -52,10 +54,7 @@
   Stop();
   SockNode().Stop();
   if (state() == eStateUnregistered) {
      SockNode().Remove();
      SockClient().Remove();
      SockServer().Remove();
      SockSub().Remove();
      for (auto &p : sockets_) { p->Remove(); }
   }
}

@@ -66,16 +65,14 @@
   } else if (nworker > 16) {
      nworker = 16;
   }

   SockNode().Start();
   ServerStart(server_cb, nworker);
   SubscribeStartWorker(sub_cb, nworker);
   ClientStartWorker(client_cb, nworker);
}
void TopicNode::Stop()
{
   SockSub().Stop();
   SockServer().Stop();
   SockClient().Stop();
   for (auto &p : sockets_) { p->Stop(); }
}

bool TopicNode::Register(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
@@ -114,7 +111,7 @@
      return sock.Send(&BHTopicCenterAddress(), head, body, onResult);
   } else {
      MsgI reply;
      DEFER1(reply.Release(shm_););
      DEFER1(reply.Release(););
      BHMsgHead reply_head;
      bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms);
      if (r) {
@@ -142,7 +139,7 @@
      return sock.Send(&BHTopicCenterAddress(), head, body);
   } else {
      MsgI reply;
      DEFER1(reply.Release(shm_););
      DEFER1(reply.Release(););
      BHMsgHead reply_head;
      bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms);
      r = r && reply_head.type() == kMsgTypeCommonReply && reply.ParseBody(reply_body);
@@ -175,7 +172,7 @@
      return sock.Send(&BHTopicCenterAddress(), head, body);
   } else {
      MsgI reply;
      DEFER1(reply.Release(shm_););
      DEFER1(reply.Release(););
      BHMsgHead reply_head;
      bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms);
      r = r && reply_head.type() == kMsgTypeCommonReply;
@@ -286,7 +283,7 @@
   return SockClient().Start(onData, nworker);
}

bool TopicNode::ClientAsyncRequest(const MsgRequestTopic &req, std::string &out_msg_id, const RequestResultCB &cb)
bool TopicNode::ClientAsyncRequest(const BHAddress &remote_addr, const MsgRequestTopic &req, std::string &out_msg_id, const RequestResultCB &cb)
{
   if (!IsOnline()) {
      SetLastError(eNotRegistered, "Not Registered.");
@@ -352,7 +349,7 @@
   }
}

bool TopicNode::ClientSyncRequest(const MsgRequestTopic &request, std::string &out_proc_id, MsgRequestTopicReply &out_reply, const int timeout_ms)
bool TopicNode::ClientSyncRequest(const BHAddress &remote_addr, const MsgRequestTopic &request, std::string &out_proc_id, MsgRequestTopicReply &out_reply, const int timeout_ms)
{
   if (!IsOnline()) {
      SetLastError(eNotRegistered, "Not Registered.");
@@ -369,7 +366,7 @@
         head.set_topic(request.topic());

         MsgI reply_msg;
         DEFER1(reply_msg.Release(shm_););
         DEFER1(reply_msg.Release(););
         BHMsgHead reply_head;

         if (sock.SendAndRecv(addr.mq_id().data(), head, request, reply_msg, reply_head, timeout_ms) &&
@@ -406,7 +403,7 @@
   AddRoute(head, sock.id());

   MsgI reply;
   DEFER1(reply.Release(shm_));
   DEFER1(reply.Release());
   BHMsgHead reply_head;

   if (sock.SendAndRecv(&BHTopicCenterAddress(), head, query, reply, reply_head, timeout_ms)) {
@@ -445,7 +442,7 @@
         return sock.Send(&BHTopicBusAddress(), head, pub);
      } else {
         MsgI reply;
         DEFER1(reply.Release(shm()););
         DEFER1(reply.Release(););
         BHMsgHead reply_head;
         MsgCommonReply reply_body;
         return sock.SendAndRecv(&BHTopicBusAddress(), head, pub, reply, reply_head, timeout_ms) &&
@@ -478,7 +475,7 @@
         return sock.Send(&BHTopicBusAddress(), head, sub);
      } else {
         MsgI reply;
         DEFER1(reply.Release(shm()););
         DEFER1(reply.Release(););
         BHMsgHead reply_head;
         return sock.SendAndRecv(&BHTopicBusAddress(), head, sub, reply, reply_head, timeout_ms) &&
                reply_head.type() == kMsgTypeCommonReply &&
@@ -518,7 +515,7 @@

   auto &sock = SockSub();
   MsgI msg;
   DEFER1(msg.Release(shm()););
   DEFER1(msg.Release(););
   BHMsgHead head;
   //TODO error msg.
   if (sock.SyncRecv(msg, head, timeout_ms) && head.type() == kMsgTypePublish) {