move ref count into msg meta, only 1 poinetr now.

lichao

2021-04-23 dc86ace85e437ecb8a2e728e4dce36d02bbb8a6e

 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)
@@ -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.");