remove AtomicQueue, not used.

lichao

2021-06-01 72851db66655912cb9c92300a80985fb9797d168

 box/center.cpp

@@ -16,7 +16,11 @@
 * =====================================================================================
 */
#include "center.h"
#include "center_topic_node.h"
#include "io_service.h"
#include "node_center.h"
#include "tcp_proxy.h"
#include "tcp_server.h"
#include <chrono>

using namespace std::chrono;
@@ -64,7 +68,7 @@
   return [&](auto &&rep_body) {
      auto reply_head(InitMsgHead(GetType(rep_body), center->id(), head.ssn_id(), head.msg_id()));
      MQInfo remote = {head.route(0).mq_id(), head.route(0).abs_addr()};
      MsgI msg;
      MsgI msg(socket.shm());
      if (msg.Make(reply_head, rep_body)) {
         DEFER1(msg.Release(););
         center->SendAllocMsg(socket, remote, msg);
@@ -72,7 +76,7 @@
   };
}

bool AddCenter(std::shared_ptr<Synced<NodeCenter>> center_ptr)
bool AddCenter(std::shared_ptr<Synced<NodeCenter>> center_ptr, SharedMemory &shm, TcpProxy &tcp_proxy)
{
   // command
   auto OnCommand = [center_ptr](ShmSocket &socket, ShmMsgQueue::RawData &cmd) -> bool {
@@ -86,13 +90,45 @@
      auto onInit = [&](const int64_t request) {
         return center->OnNodeInit(socket, request);
      };
      BHCenterHandleInit(onInit);
      BHCenterHandleInit(socket.shm(), onInit);
      center->OnTimer();
   };

   auto OnCenter = [=](ShmSocket &socket, MsgI &msg, BHMsgHead &head) -> bool {
   auto OnCenter = [=, &tcp_proxy](ShmSocket &socket, MsgI &msg, BHMsgHead &head) -> bool {
      auto &center = *center_ptr;
      auto replyer = MakeReplyer(socket, head, center);

      if (!head.dest().ip().empty()) { // other host, proxy
         auto valid = [&]() { return head.route_size() == 1; };
         if (!valid()) { return false; }

         if (head.type() == kMsgTypeRequestTopic) {
            typedef MsgRequestTopicReply Reply;
            Reply reply;
            if (!center->CheckMsg(head, reply)) {
               replyer(reply);
            } else {
               auto onResult = [&center](BHMsgHead &head, std::string body_content) {
                  if (head.route_size() > 0) {
                     auto &back = head.route(head.route_size() - 1);
                     MQInfo dest = {back.mq_id(), back.abs_addr()};
                     head.mutable_route()->RemoveLast();
                     center->PassRemoteReplyToLocal(dest, head, std::move(body_content));
                  }
               };
               if (!tcp_proxy.Request(head.dest().ip(), head.dest().port(), msg.content(), onResult)) {
                  replyer(MakeReply<Reply>(eError, "send request failed."));
               } else {
                  // success
               }
            }
            return true;
         } else {
            // ignore other msgs for now.
         }
         return false;
      }

      switch (head.type()) {
         CASE_ON_MSG_TYPE(ProcInit);
         CASE_ON_MSG_TYPE(Register);
@@ -105,7 +141,7 @@
      default: return false;
      }
   };
   BHCenter::Install("#center.main", OnCenter, OnCommand, OnCenterIdle, BHTopicCenterAddress(), 1000);
   BHCenter::Install("@center.main", OnCenter, OnCommand, OnCenterIdle, BHTopicCenterAddress(shm), 1000);

   auto OnBusIdle = [=](ShmSocket &socket) {};
   auto OnBusCmd = [=](ShmSocket &socket, ShmMsgQueue::RawData &val) { return false; };
@@ -123,20 +159,14 @@
         } else {
            replyer(MakeReply(eSuccess));
            if (clients.empty()) { return; }

            auto it = clients.begin();
            do {
               auto &cli = *it;
            for (auto &cli : clients) {
               auto node = cli.weak_node_.lock();
               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_id_, cli.mq_abs_addr_}, msg);
                  ++it;
               } else {
                  it = clients.erase(it);
               }
            } while (it != clients.end());
            }
         }
      };
      switch (head.type()) {
@@ -147,7 +177,7 @@
      }
   };

   BHCenter::Install("#center.bus", OnPubSub, OnBusCmd, OnBusIdle, BHTopicBusAddress(), 1000);
   BHCenter::Install("@center.bus", OnPubSub, OnBusCmd, OnBusIdle, BHTopicBusAddress(shm), 1000);

   return true;
}
@@ -170,22 +200,23 @@

BHCenter::BHCenter(Socket::Shm &shm)
{
   auto gc = [&](const MQId id) {
      auto r = ShmSocket::Remove(shm, id);
      if (r) {
         LOG_DEBUG() << "remove mq " << id << " ok\n";
      }
   };

   auto nsec = NodeTimeoutSec();
   auto center_ptr = std::make_shared<Synced<NodeCenter>>("#bhome_center", gc, nsec, nsec * 3); // *3 to allow other clients to finish sending msgs.
   AddCenter(center_ptr);
   auto center_ptr = std::make_shared<Synced<NodeCenter>>("@bhome_center", nsec, nsec * 3); // *3 to allow other clients to finish sending msgs.
   io_service_.reset(new IoService);
   tcp_proxy_.reset(new TcpProxy(io_service_->io()));

   AddCenter(center_ptr, shm, *tcp_proxy_);

   for (auto &kv : Centers()) {
      auto &info = kv.second;
      sockets_[info.name_] = std::make_shared<ShmSocket>(info.mq_.offset_, shm, info.mq_.id_);
   }

   topic_node_.reset(new CenterTopicNode(center_ptr, shm));
   tcp_server_.reset(new TcpServer(io_service_->io(), kBHCenterPort, center_ptr));
}

BHCenter::~BHCenter() { Stop(); }

bool BHCenter::Start()
{
@@ -193,12 +224,16 @@
      auto &info = kv.second;
      sockets_[info.name_]->Start(1, info.handler_, info.raw_handler_, info.idle_);
   }

   topic_node_->Start();
   return true;
}

bool BHCenter::Stop()
{
   tcp_proxy_.reset();
   tcp_server_.reset();
   io_service_.reset();
   topic_node_->Stop();
   for (auto &kv : sockets_) {
      kv.second->Stop();
   }