refactor, start tcp pub/sub.

lichao

2021-06-23 c1e39e20ca42b21eeac8b5068fa1f921bf9a070f

 utest/tcp_test.cpp

@@ -16,7 +16,10 @@
 * =====================================================================================
 */

#include "defs.h"
#include "node_center.h"
#include "tcp_connection.h"
#include "tcp_proxy.h"
#include "tcp_server.h"
#include "util.h"
#include <sys/ioctl.h>
@@ -31,21 +34,55 @@

BOOST_AUTO_TEST_CASE(TcpTest)
{
   const std::string bind_addr = "127.0.0.1";
   const std::string connect_addr = "127.0.0.1";
   const uint16_t port = 10000;
   const uint16_t port = kBHCenterPort;

   TcpServer server(port);
   server.Start();

   boost::asio::io_context io;
   IoService io;

   tcp::endpoint dest(ip::address::from_string(connect_addr), port);
   for (int i = 0; i < 10; ++i) {
      TcpRequest1::Create(io, dest, "client->server " + std::to_string(i));
   }

   io.run();
   auto NewRequest = [&]() {
      MsgRequestTopic req;
      req.set_topic("#center_query_procs");
      req.set_data("");
      auto head = InitMsgHead(GetType(req), "#test_proc", 1000000);
      auto route = head.add_route();
      route->set_mq_id(12345);
      route->set_abs_addr(67890);
      head.set_topic(req.topic());

      head.mutable_dest()->set_ip(connect_addr);
      // head.mutable_dest()->set_port(port);
      // head.mutable_dest()->set_mq_id(201);
      // head.mutable_dest()->set_abs_addr(10072);

      return (MsgI::Serialize(head, req));
   };
   auto onReply = [](BHMsgHead &head, std::string body_content) {
      static int n = 0;
      printf("reply %d: ", ++n);
      MsgRequestTopicReply reply;
      if (reply.ParseFromString(body_content)) {
         if (IsSuccess(reply.errmsg().errcode())) {
            printf("\ncontent: %s\n", reply.data().c_str());
         } else {
            printf("error: %s\n", reply.errmsg().errstring().c_str());
         }
      } else {
         printf("parse error\n");
      }
   };
   for (int i = 0; i < 100; ++i) {
      auto request = NewRequest();
      TcpRequest1::Create(io.io(), dest, request, onReply);
   }
   Sleep(2s);
   printf("-------------------------------------------------------\n");
   for (int i = 0; i < 3; ++i) {
      auto request = NewRequest();
      TcpRequest1::Create(io.io(), dest, request, onReply);
   }
   Sleep(2s);

   printf("TcpTest\n");
}