support tcp pub/sub.

lichao

2021-06-30 ae17d1439b35b55212c3a30712e0a60b1d6a99c0

 src/bh_api.cc

@@ -1,9 +1,11 @@
#include "bh_api.h"
#include "defs.h"
#include "topic_node.h"
#include <chrono>
#include <cstdio>
#include <memory>

using namespace std::chrono_literals;
using namespace bhome_shm;
using namespace bhome_msg;

@@ -30,20 +32,33 @@
}
std::unique_ptr<TopicNode> &ProcNodePtr()
{
   // client side init here.
   static std::mutex mtx;
   std::lock_guard<std::mutex> lk(mtx);
   auto InitLog = []() {
      ns_log::AddLog(BHLogDir() + "bhshmq_node_" + GetProcExe() + ".log", true);
      return true;
   };
   static bool init_log = InitLog();

   static std::string shm_name;
   static std::unique_ptr<TopicNode> ptr;
   if (!ptr && GlobalInit(BHomeShm())) {
      auto InitLog = []() {
         auto id = GetProcExe();
         char path[200] = {0};
         sprintf(path, "/opt/vasystem/valog/bhshmq_node_%s.log", id.c_str());
         ns_log::AddLog(path);
         return true;
      };
      static bool init_log = InitLog();
      ptr.reset(new TopicNode(BHomeShm()));

   std::lock_guard<std::mutex> lk(mtx);
   if (shm_name != BHomeShmName()) {
      shm_name = BHomeShmName();
      LOG_INFO() << "using shm " << shm_name;

      ptr.reset();
      // must reset/stop node before call BHomeShm() which resets shm.

      auto &shm = BHomeShm();
      for (int i = 0; !ptr && i < 3; ++i) {
         if (GlobalInit(shm)) {
            ptr.reset(new TopicNode(shm));
         } else {
            std::this_thread::sleep_for(1s); // make sure shm init done.
         }
      }
   }
   return ptr;
}
@@ -211,6 +226,10 @@
{
   return BHApi_In1_Out1<MsgTopicList>(&TopicNode::Subscribe, topics, topics_len, reply, reply_len, timeout_ms);
}
int BHSubscribeNetTopics(const void *topics, const int topics_len, void **reply, int *reply_len, const int timeout_ms)
{
   return BHApi_In1_Out1<MsgTopicList>(&TopicNode::SubscribeNet, topics, topics_len, reply, reply_len, timeout_ms);
}

int BHPublish(const void *msgpub,
              const int msgpub_len,