/**
 * @file clientDemo.cpp
 * @author root
 * @brief 简要描述文件的功能或用途
 * @date 25-7-16
 *
 * 详细的文件描述或注意事项可以放在这里。
 */
#include "clientDemo.h"
#include "bareMessageDefine.h"
#include "XeleSecuritiesUserApiStruct_Bare.h"

#include <errno.h>
#include <cstring>
#include <fcntl.h>
#include <sys/select.h>

/// Manager响应报文处理接口配置
static ResponseHandlerItem g_managerHandlers[] = {
    {MID_RSP_USER_LOGIN, new RspManagerUserLoginHandle()},
    {MID_RSP_USER_LOGOUT, new RspManagerUserLogoutHandle()}
};

/// 柜台响应报文处理接口配置
static ResponseHandlerItem g_counterHandlers[] = {
    {TID_RSP_USER_LOGIN, new RspUserLoginHandler()},
    {TID_RSP_USER_LOGOUT, new RspCounterUserLogoutHandle()},
    {TID_RSP_INIT_TRADER, new RspInitTraderHandler()},
    {TID_RSP_ORDER_INSERT, new RspOrderInsertHandler()},
    {TID_ERR_ORDER_INSERT, new ErrOrderInsertHandler()},
    {TID_RTN_ORDER, new RtnOrderHandler()},
    {TID_RTN_TRADE, new RtnTradeHandler()},
    {TID_RSP_ORDER_ACTION, new RspOrderActionHandler()},
    {TID_ERR_ORDER_ACTION, new ErrOrderActionHandler()},
    {TID_RSP_QRY_ORDER, new RspOrderQueryHandler()}
};

ClientDemo::ClientDemo() {
  /// epoll_create(max_size), max_size is invalid after linux2.6.8, but must grater than 0
  m_epfd = epoll_create(64);

  m_isRunning = false;
  m_managerHeartbeatStart = false;
  m_counterQueryHeartbeatStart = false;
  m_counterTradeHeartbeatStart = false;

  m_managerLogin = false;
  m_counterQueryLogin = false;
  m_counterTradeLogin = false;

  m_timoutCountManager = 0;
  m_timoutCountCounterQuery = 0;
  m_timoutCountCounterTrade = 0;

  init();
};

ClientDemo::~ClientDemo() {
  PRINT_INFO("~ClientDemo");
  m_isRunning = false;

  if (m_heatBtThread.joinable()) {
    m_heatBtThread.join();
  }

  if (m_mainEvThread.joinable()) {
    m_mainEvThread.join();
  }

  close(m_epfd);
}

void ClientDemo::init() {
  /// 注册响应报文处理接口
  for (int i = 0; i < sizeof(g_managerHandlers)/sizeof(ResponseHandlerItem); ++i) {
    if (g_managerHandlers[i].handler == nullptr) {
      return;
    }

    m_managerHandlers[g_managerHandlers[i].messageID] = g_managerHandlers[i].handler;

  }

  for (int i = 0; i < sizeof(g_counterHandlers)/sizeof(ResponseHandlerItem); ++i) {
    if (g_counterHandlers[i].handler == nullptr) {
      return;
    }

    m_counterHandlers[g_counterHandlers[i].messageID] = g_counterHandlers[i].handler;

  }
}

int ClientDemo::connectManager() {
  auto socket = connect(GlobalDataManager::GetInstance().getManagerPort(),
                        GlobalDataManager::GetInstance().getManagerIp(), TcpType::ManagerTcp);
  m_socketMap.emplace(TcpType::ManagerTcp, socket);
  return socket;
}

int ClientDemo::connectCounterQuery() {
  auto socket = connect(GlobalDataManager::GetInstance().getCounterQueryPort(),
                        GlobalDataManager::GetInstance().getCounterQueryIp(), TcpType::CounterQueryTcp);
  m_socketMap.emplace(TcpType::CounterQueryTcp, socket);
  return socket;
}

int ClientDemo::connectCounterTrade() {
  auto socket = connect(GlobalDataManager::GetInstance().getCounterTradePort(),
                        GlobalDataManager::GetInstance().getCounterTradeIp(), TcpType::CounterTradeTcp);
  m_socketMap.emplace(TcpType::CounterTradeTcp, socket);
  return socket;
}

int ClientDemo::connect(uint16_t port, std::string ip, TcpType type) {
  /// Connect TcpServer
  struct sockaddr_in servAddr{};
  /// Not Use AF_INET, AF_INET Is For servAddr.sin_family.
  auto m_connFd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
  if (m_connFd == -1) {
    PRINT_INFO("Create Socket Error! [%s]", strerror(errno));
    return 0;
  }

  memset(&servAddr, 0, sizeof(servAddr));
  servAddr.sin_family = AF_INET;
  servAddr.sin_addr.s_addr = inet_addr(ip.c_str());
  servAddr.sin_port = htons(port);

  /// Set NonBlock Mode, Epoll ET Mode Only Support NoBlocking Socket.
  int flags = fcntl(m_connFd, F_GETFL, 0);
  fcntl(m_connFd, F_SETFL, flags|O_NONBLOCK);

  if (connectNoBlocking(m_connFd, (struct sockaddr*)&servAddr, sizeof(servAddr)) != 0) {
    PRINT_INFO("Connect Tcp Server Error! socektType %d", type);
    return 0;
  }

  /// 交易链路socket设置 EPOLLIN 事件， ET模式， 事件到达仅触发一次
  struct epoll_event ev{};
  ev.events = EPOLLIN | EPOLLOUT | EPOLLPRI | EPOLLHUP | EPOLLERR | EPOLLET;
  ev.data.fd = m_connFd;
  SendEvent sendEvent{};
  sendEvent.socket = m_connFd;
  sendEvent.socketType = type;
  m_socketEventMap.emplace(m_connFd, sendEvent);
  auto ret = epoll_ctl(m_epfd, EPOLL_CTL_ADD, m_connFd, &ev);
  if (0 != ret) {
    close(m_epfd);
    PRINT_INFO("epoll_ctl fail, errno: [%d], errmsg: [%s]",errno, strerror(errno));
  }

  return m_connFd;
}

int ClientDemo::connectNoBlocking(int fd, struct sockaddr *addr, socklen_t addrLen) {
  for (;;) {
    int ret = ::connect(fd, addr, addrLen);
    // 因为fd被设置成了非阻塞, 大部分情况下, 都是返回-1, errno被设置成EINPROGRESS, 表示正在连接中
    if (ret == 0) {
      return 0;
    } else if (ret == -1) {
      if (errno == EINTR) {
        /// connect Interrupt By Signal, Try Again
        continue;
      } else if (errno == EINPROGRESS) {
        /// Connecting...
        break;
      } else {
        /// Error
        return -1;
      }
    }
  }

  /// 三次握手过程中, fd会有发送Sync的处理, 这里用select来判断是否有发出消息的事件
  fd_set writeset;
  FD_ZERO(&writeset);
  FD_SET(fd, &writeset);

  /// Wait 3 seconds
  struct timeval tv{};
  tv.tv_sec = 3;
  tv.tv_usec = 0;
  if (select(fd+1, nullptr, &writeset, nullptr, &tv) != 1) {
    /// Error
    return -1;
  }

  /// 判断一下fd的状态， 以确保是成功连接的状态
  int err;
  auto len = (socklen_t)(sizeof(err));
  if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0) {
    return -1;
  }

  return err;
}

int ClientDemo::run() {
  /// 先设置运行状态
  m_isRunning = true;

  /// 启动主要逻辑处理线程和心跳线程, 后续可以考虑加上OrderWarm机制的处理
  m_mainEvThread = std::thread(&ClientDemo::eventLoop, this);
  m_heatBtThread = std::thread(&ClientDemo::heartBeatThread, this);

  return 0;
}

void ClientDemo::heartBeatThread() {
  m_sndHeatBtCtManager = 1;
  m_sndHeatBtCtCounterQuery = 1;
  m_sndHeatBtCtCounterTrade = 1;
  /// manager
  memset(m_sHeartBtBufManager, 0, sizeof(m_sHeartBtBufManager));

  /// 构造 交易客户端发往交易服务端 的心跳报文
  auto pReqHeadManager = (CXeleFairReqHeadField*)m_sHeartBtBufManager;
  pReqHeadManager->ProtocolID = PROTOCOL_ID;
  pReqHeadManager->MessageID = MID_HEAETBEAT_CLIENT_2_SERVER;
  pReqHeadManager->MessageLength = sizeof(CXeleFairReqHeadField) + sizeof(CXeleHeartBeatManager);
  pReqHeadManager->SessionID = GlobalDataManager::GetInstance().getManagerSessionId();
  pReqHeadManager->Token = GlobalDataManager::GetInstance().getManagerToken();

  /// counter query
  memset(m_sHeartBtBufCounterQuery, 0, sizeof(m_sHeartBtBufCounterQuery));

  /// 构造 交易客户端发往交易服务端 的心跳报文
  auto pReqHeadCounterQuery = (CXeleFairReqHeadField*)m_sHeartBtBufCounterQuery;
  pReqHeadCounterQuery->ProtocolID = PROTOCOL_ID;
  pReqHeadCounterQuery->MessageID = TID_HEAETBEAT_QUERY_CLIENT_2_SERVER;
  pReqHeadCounterQuery->MessageLength = sizeof(CXeleFairReqHeadField) + sizeof(CXeleHeartBeat);
  pReqHeadCounterQuery->SessionID = GlobalDataManager::GetInstance().getCounterSessionId();
  pReqHeadCounterQuery->Token = GlobalDataManager::GetInstance().getCounterToken();

  /// counter trade
  memset(m_sHeartBtBufCounterTrade, 0, sizeof(m_sHeartBtBufCounterTrade));

  /// 构造交易客户端发往交易服务端 的心跳报文
  auto pReqHeadCounterTrade = (CXeleFairReqHeadField*)m_sHeartBtBufCounterTrade;
  pReqHeadCounterTrade->ProtocolID = PROTOCOL_ID;
  pReqHeadCounterTrade->MessageID = TID_HEARTBEAT_TRADE_CLIENT_2_SERVER;
  pReqHeadCounterTrade->MessageLength = sizeof(CXeleFairReqHeadField) + sizeof(CXeleHeartBeat);
  pReqHeadCounterTrade->SessionID = GlobalDataManager::GetInstance().getCounterSessionId();
  pReqHeadCounterTrade->Token = GlobalDataManager::GetInstance().getCounterToken();

  struct epoll_event ev{};

  PRINT_INFO("heartbeat thread start...");
  while (m_isRunning) {
    /// manager heartbeat
    if (m_managerHeartbeatStart) {
      if (m_sndHeatBtCtManager > GlobalDataManager::GetInstance().getManagerHeartBeatInterval()) {
        /// Set Send HeartBeat Event
        /// 注册心跳发送事件
        auto fd = m_socketMap[TcpType::ManagerTcp];
        SendEvent sendEvent{};
        sendEvent.socket = fd;
        sendEvent.socketType = TcpType::ManagerTcp;
        sendEvent.messageID = MID_HEAETBEAT_CLIENT_2_SERVER;
        sendEvent.messageLength = pReqHeadManager->MessageLength;
        m_socketEventMap[fd] = sendEvent;
        /// 注册EPOLLOUT事件
        ev.events = EPOLLOUT;
        ev.data.fd = fd;
        epoll_ctl(m_epfd, EPOLL_CTL_MOD, fd, &ev);

        m_sndHeatBtCtManager = 1;
      }

      /// 交易服务端没有发送心跳报文, 超时处理
      if (m_timoutCountManager >= GlobalDataManager::GetInstance().getManagerHeartBeatTimeout()) {
        /// TimeOut
        PRINT_INFO("Manager HeartBeat Timeout!");
        m_isRunning = false;
      }

      m_sndHeatBtCtManager++;
      m_timoutCountManager++;
    }
    /// counter heartbeat
    if (m_counterQueryHeartbeatStart) {
      if (m_sndHeatBtCtCounterQuery > GlobalDataManager::GetInstance().getCounterHeartbeatInterval()) {
        /// Set Send HeartBeat Event
        /// 注册心跳发送事件
        auto fd = m_socketMap[TcpType::CounterQueryTcp];
        SendEvent sendEvent{};
        sendEvent.socket = fd;
        sendEvent.socketType = TcpType::CounterQueryTcp;
        sendEvent.messageID = TID_HEAETBEAT_QUERY_CLIENT_2_SERVER;
        sendEvent.messageLength = pReqHeadCounterQuery->MessageLength;
        m_socketEventMap[fd] = sendEvent;
        /// 注册EPOLLOUT事件
        ev.events = EPOLLOUT;
        ev.data.fd = fd;
        epoll_ctl(m_epfd, EPOLL_CTL_MOD, fd, &ev);

        m_sndHeatBtCtCounterQuery = 1;
      }

      /// 交易服务端没有发送心跳报文, 超时处理
      if (m_timoutCountCounterQuery >= GlobalDataManager::GetInstance().getCounterHeartbeatTimeout()) {
        /// TimeOut
        PRINT_INFO("Counter Query HeartBeat Timeout!");
        m_isRunning = false;
      }

      m_sndHeatBtCtCounterQuery++;
      m_timoutCountCounterQuery++;
    }

    if (m_counterTradeHeartbeatStart) {
      if (m_sndHeatBtCtCounterTrade > GlobalDataManager::GetInstance().getCounterHeartbeatInterval()) {
        /// Set Send HeartBeat Event
        /// 注册心跳发送事件
        auto fd = m_socketMap[TcpType::CounterTradeTcp];
        SendEvent sendEvent{};
        sendEvent.socket = fd;
        sendEvent.socketType = TcpType::CounterTradeTcp;
        sendEvent.messageID = TID_HEARTBEAT_TRADE_CLIENT_2_SERVER;
        sendEvent.messageLength = pReqHeadCounterTrade->MessageLength;
        m_socketEventMap[fd] = sendEvent;

        /// 注册EPOLLOUT事件
        ev.events = EPOLLOUT;
        ev.data.fd = fd;
        epoll_ctl(m_epfd, EPOLL_CTL_MOD, fd, &ev);

        m_sndHeatBtCtCounterTrade = 1;
      }

      /// 交易服务端没有发送心跳报文, 超时处理
      if (m_timoutCountCounterTrade >= GlobalDataManager::GetInstance().getCounterHeartbeatTimeout()) {
        /// TimeOut
        PRINT_INFO("Counter Trade HeartBeat Timeout!");
        m_isRunning = false;
      }

      m_sndHeatBtCtCounterTrade++;
      m_timoutCountCounterTrade++;
    }
    sleep(1);
  }
  PRINT_INFO("heartbeat thread end...");
}

void ClientDemo::eventLoop() {
  int nEvent;
  size_t readLen = 0;
  struct epoll_event ev{}, events[128]{};

  PRINT_INFO("eventLoop thread start...");
  while (m_isRunning) {
    /// epoll_wait 等到事件到来, 超时时间设置为500ms
    nEvent = epoll_wait(m_epfd, events, sizeof(events)/sizeof(struct epoll_event), 500);

    /// 事件处理循环
    for (int i = 0; i < nEvent; ++i) {
      int event = events[i].events;
      int fd = events[i].data.fd;
      if (m_socketEventMap.find(fd) == m_socketEventMap.end()) {
        continue;
      }

      // 错误检测优先
      if (event & EPOLLERR) {
        int error = 0;
        socklen_t len = sizeof(error);
        getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &len);
        PRINT_INFO("socket %d error, reaseon: %s", fd, strerror(error));
        close(fd);
        continue;
      }

      // 对端断链检测
      if (event & (EPOLLRDHUP | EPOLLHUP)) {
        // 1. 取消 epoll 监听
        epoll_ctl(m_epfd, EPOLL_CTL_DEL, fd, nullptr);
        // 2. 获取对端关闭原因
        int error = 0;
        socklen_t len = sizeof(error);
        getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &len);
        // 3. 关闭 socket
        shutdown(fd, SHUT_RDWR);
        close(fd);
        // 4. 清理相关资源
        if (m_socketEventMap.find(fd) != m_socketEventMap.end()) {
          m_socketEventMap[fd].messageID = 0;
          m_socketEventMap[fd].messageLength = 0;
        }
        // 记录日志
        PRINT_INFO("Connection closed by peer (reason: %s)", strerror(error));
        continue;
      }

      if (event & EPOLLIN) {
        /// 处理接收事件
        readLen = recvMessage(fd);
        if (readLen <= 0) {
          PRINT_INFO("socket %d close", fd);
          m_isRunning = false;
        } else {
          doResponse(events[i]);
        }
      } else if (event & EPOLLOUT) {
        /// 处理发送事件
        auto sndEv = &m_socketEventMap.at(fd);
        handleSendEvent(sndEv);
        sndEv->messageID = 0;
        sndEv->messageLength = 0;
        events[i].events = EPOLLIN;
        if (epoll_ctl(m_epfd, EPOLL_CTL_MOD, fd, &events[i]) == -1) {
          PRINT_INFO("epoll_ctl update ptr failed");
        }
      } else {
        PRINT_INFO("socket %d close", fd);
        if (epoll_ctl(m_epfd, EPOLL_CTL_DEL, fd, &events[i]) == -1) {
          PRINT_INFO("epoll_ctl update ptr failed");
        }
      }
    }
  }
  PRINT_INFO("eventLoop thread end...");
}

void ClientDemo::handleSendEvent(SendEvent *sndEvent) {
  ssize_t sendLen = 0;
  if (sndEvent->messageID == 0 && sndEvent->messageLength == 0) {
    return;
  }
  char* sendPtr = nullptr;
  if (sndEvent->socketType == TcpType::ManagerTcp) {
    if (sndEvent->messageID == MID_HEAETBEAT_CLIENT_2_SERVER) {
      sendPtr = m_sHeartBtBufManager;
    } else {
      sendPtr = m_sendBufManager;
    }
  } else if (sndEvent->socketType == TcpType::CounterQueryTcp) {
    if (sndEvent->messageID == TID_HEAETBEAT_QUERY_CLIENT_2_SERVER) {
      sendPtr = m_sHeartBtBufCounterQuery;
    } else {
      sendPtr = m_sendBufCounterQuery;
    }
  } else if (sndEvent->socketType == TcpType::CounterTradeTcp) {
    if (sndEvent->messageID == TID_HEARTBEAT_TRADE_CLIENT_2_SERVER) {
      sendPtr = m_sHeartBtBufCounterTrade;
    } else {
      sendPtr = m_sendBufCounterTrade;
    }
  } else {
    PRINT_INFO("other socketType %d", sndEvent->socketType);
  }
  if (sendPtr) {
    sendLen = write(sndEvent->socket, sendPtr, sndEvent->messageLength);
    if (sendLen != sndEvent->messageLength) {
      /// todo: need optimize
      PRINT_INFO("Send socket %d, socketType %d, messageId %d, messageLenght %d Failed!",
                 sndEvent->socket,
                 sndEvent->socketType,
                 sndEvent->messageID,
                 sndEvent->messageLength);
    } else {
      PRINT_INFO("Send socket %d, socketType %d, messageId %d, messageLenght %d Success!",
                 sndEvent->socket,
                 sndEvent->socketType,
                 sndEvent->messageID,
                 sndEvent->messageLength);
    }
  }
}

/// 相对可靠的读取数据接口封装
size_t ClientDemo::readn(int fd, void *buf, size_t n) {
  ssize_t nread = 0;
  size_t nleft = n;
  char *pBuf = (char*)buf;

  do {
    nread = read(fd, pBuf, nleft);

    if (nread == -1) {
      if (errno == EINTR ||
          errno == EAGAIN) {
        continue;
      } else {
        return 0;
      }
    } else if (nread == 0) {
      break;
    }

    nleft -= nread;
    pBuf += nread;
  } while (nleft);

  return (n - nleft);
}

/// 读取回应数据到m_recvBuf, 先读取Head, 再读取Body, 指定字节数, 防止粘包
size_t ClientDemo::recvMessage(int fd) {
  size_t readHeadLen = 0, readBodyLen = 0;

  memset(m_recvBuf, 0, sizeof(m_recvBuf));
  readHeadLen = readn(fd, m_recvBuf, sizeof(CXeleFairRspHeadField));
  if (readHeadLen <= 0) {
    return readHeadLen;
  }

  /// Parse Body Len
  auto rspHead = (CXeleFairRspHeadField*)m_recvBuf;
  size_t bodyLen = rspHead->MessageLength - sizeof(CXeleFairRspHeadField);

  if (bodyLen > 0) {
    readBodyLen = readn(fd, m_recvBuf+sizeof(CXeleFairRspHeadField), bodyLen);
    if (readBodyLen <= 0) {
      return readBodyLen;
    }
  }

  return (readHeadLen+readBodyLen);
}

int ClientDemo::doResponse(epoll_event event) {
  /// 收到消息后，根据fd找到对应的socket类型（manager消息或者counter消息），找到类型后，进行相应的消息处理
  if (m_socketEventMap.find(event.data.fd) == m_socketEventMap.end()) {
    return 0;
  }
  auto sendEvent = m_socketEventMap.at(event.data.fd);
  if (sendEvent.socketType == TcpType::ManagerTcp) {
    /// manager消息处理
    auto pRspHead = (CXeleFairRspHeadField*)m_recvBuf;
    /// Reset timout
    /// 当前链路存在消息，就表示链路正常，将心跳时间置0
    m_timoutCountManager = 0;

    if (pRspHead->MessageID == MID_HEARTBEAT_SERVER_2_CLIENT) {
      /// HeartBeat
      PRINT_INFO("====================HeartBeat_Manager2Client====================");
      return 0;
    }

    if (m_managerHandlers.find(pRspHead->MessageID) == m_managerHandlers.end()) {
      PRINT_INFO("UnSupport MessageID [%d]", pRspHead->MessageID);
      return -1;
    }

    responseHandler *pHandler = m_managerHandlers[pRspHead->MessageID];
    if (pHandler == nullptr) {
      PRINT_INFO("MessageId %d Handler Is Null!", pRspHead->MessageID);
      return -1;
    }
    /// 根据回报的MessageId来处理回报应答消息
    auto errorCode = pHandler->doResponse(m_recvBuf);
    /// 当manager的登录答应成功时，可以开始登陆柜台的查询链路
    if (pRspHead->MessageID == MID_RSP_USER_LOGIN && errorCode == SHARE_NO_ERROR) {
      /// 登录成功后开始发送心跳
      PRINT_INFO("start manager heartbeat...");
      m_managerHeartbeatStart = true;
      m_managerLogin = true;
      ///先连接tcp
      PRINT_INFO("******************start connect counter query tcp******************");
      auto fd = connectCounterQuery();
      if (fd > 0) {
        /// 在发送登录消息
        PRINT_INFO("send login message to counter query tcp");
        TestCounterLogin();
      }
    } else if (pRspHead->MessageID == MID_RSP_USER_LOGOUT && errorCode == SHARE_NO_ERROR) {
      PRINT_INFO("stop manager heartbeat...");
      m_managerHeartbeatStart = false;
      m_managerLogin = false;
    }
    return errorCode;
  } else {
    /// counter消息处理
    auto pRspHead = (CXeleFairRspHeadField*)m_recvBuf;

    /// Reset timout
    /// 当前链路存在消息，就表示链路正常，将心跳时间置0
    if (sendEvent.socketType == TcpType::CounterQueryTcp) {
      m_timoutCountCounterQuery = 0;
    } else if (sendEvent.socketType == TcpType::CounterTradeTcp) {
      m_timoutCountCounterTrade = 0;
    }
    if (pRspHead->MessageID == TID_HEARTBEAT_TRADE_SERVER_2_CLIENT) {
      /// HeartBeat
      PRINT_INFO("====================HeartBeat_TradeServer2Client====================");
      return 0;
    } else if (pRspHead->MessageID == TID_HEARTBEAT_QUERY_SERVER_2_CLIENT) {
      /// HeartBeat
      PRINT_INFO("====================HeartBeat_QueryServer2Client====================");
      return 0;
    } else {
      if (m_counterHandlers.find(pRspHead->MessageID) == m_counterHandlers.end()) {
        PRINT_INFO("UnSupport MessageID [%d]", pRspHead->MessageID);
        return -1;
      }

      responseHandler *pHandler = m_counterHandlers[pRspHead->MessageID];
      if (pHandler == nullptr) {
        PRINT_INFO("MessageId %d Handler Is Null!", pRspHead->MessageID);
        return -1;
      }
      /// 根据回报的MessageId来处理回报应答消息
      auto errorCode = pHandler->doResponse(m_recvBuf);
      /// 当柜台的查询应答成功时，可以开始连接柜台的交易链路
      if (pRspHead->MessageID == TID_RSP_USER_LOGIN && errorCode == SHARE_NO_ERROR) {
        /// heartbeat
        PRINT_INFO("start counter query heartbeat...");
        m_counterQueryHeartbeatStart = true;
        m_counterQueryLogin = true;

        /// 先连接tcp
        PRINT_INFO("******************start connect counter query tcp******************");
        auto fd = connectCounterTrade();
        if (fd > 0) {
          /// 再进行交易链路的登录校验
          PRINT_INFO("send login message 2 counter query tcp");
          TestInitTrade();
        }
      } else if (pRspHead->MessageID == TID_RSP_INIT_TRADER && errorCode == SHARE_NO_ERROR) {
        /// heartbeat
        PRINT_INFO("start counter trade heartbeat...");
        m_counterTradeHeartbeatStart = true;
        m_counterTradeLogin = true;
      } else if (pRspHead->MessageID == TID_RSP_USER_LOGOUT && errorCode == SHARE_NO_ERROR) {
        /// heartbeat
        PRINT_INFO("stop counter query and trade heartbeat...");
        m_counterQueryLogin = false;
        m_counterTradeLogin = false;
        m_counterQueryHeartbeatStart = false;
        m_counterTradeHeartbeatStart = false;
      }
      return errorCode;
    }
  }
}

void ClientDemo::createSendEvent(int messageId, uint32_t messageLen, int socketType) {
  /// Set EPOLL Out Event
  auto fd = m_socketMap.at(socketType);
  if (m_socketEventMap.find(fd) !=  m_socketEventMap.end()) {
    m_socketEventMap[fd].messageID = messageId;
    m_socketEventMap[fd].messageLength = messageLen;

    struct epoll_event ev{};
    ev.events = EPOLLOUT;
    ev.data.fd = fd;
    if (epoll_ctl(m_epfd, EPOLL_CTL_MOD, fd, &ev) == -1) {
      PRINT_INFO("epoll_ctl fail");
    }
  } else {
    PRINT_INFO("fd %d not in m_socketEventMap",fd);
  }
}

void ClientDemo::createFairHead(char *buf, TcpType tcpType, uint8_t messageId, uint16_t messageLen) {
  auto pReqHead = (CXeleFairReqHeadField*) buf;
  /// Fixed 0x5f
  pReqHead->ProtocolID = PROTOCOL_ID;
  /// ReqOrderInsert Tid
  pReqHead->MessageID = messageId;
  /// Whole Packet Size, ReqHead + ReqBody
  pReqHead->MessageLength = messageLen;
  /// 登录前sessionId的值为无效默认值，后续登录成功后，才需要携带正确的值
  if (tcpType == TcpType::ManagerTcp) {
    pReqHead->SessionID = GlobalDataManager::GetInstance().getManagerSessionId();
    pReqHead->Token = GlobalDataManager::GetInstance().getManagerToken();
    pReqHead->SeqNo = GlobalDataManager::GetInstance().getSeqNoManager();
  } else {
    pReqHead->SessionID = GlobalDataManager::GetInstance().getCounterSessionId();
    pReqHead->Token = GlobalDataManager::GetInstance().getCounterToken();
    pReqHead->SeqNo = GlobalDataManager::GetInstance().getSeqNoCounterQuery();
  }
  pReqHead->RequestID = GlobalDataManager::GetInstance().requestId();
}

void ClientDemo::TestManagerLogin() {
  memset(m_sendBufManager, 0, sizeof(m_sendBufManager));
  auto MessageLength = sizeof(CXeleFairReqHeadField) + sizeof(CXeleReqUserLoginManagerField) + sizeof(SysLocalInfoField)
      + sizeof(ExternInfoField) + sizeof(AllSuperviseInfoField);
  createFairHead(m_sendBufManager, TcpType::ManagerTcp, MID_REQ_USER_LOGIN, MessageLength);

  auto pReqBody = (CXeleReqUserLoginManagerField*) (m_sendBufManager + sizeof(CXeleFairReqHeadField));
  memcpy(pReqBody->AccountID,
         GlobalDataManager::GetInstance().getAccountInfo().accountId.c_str(),
         sizeof(pReqBody->AccountID) > GlobalDataManager::GetInstance().getAccountInfo().accountId.length()
         ? GlobalDataManager::GetInstance().getAccountInfo().accountId.length() : sizeof(pReqBody->AccountID));
  memcpy(pReqBody->Password,
         GlobalDataManager::GetInstance().getAccountInfo().password.c_str(),
         sizeof(pReqBody->Password) > GlobalDataManager::GetInstance().getAccountInfo().password.length()
         ? GlobalDataManager::GetInstance().getAccountInfo().password.length() : sizeof(pReqBody->Password));

  memcpy(pReqBody->AppID,
         GlobalDataManager::GetInstance().getAppID().c_str(),
         sizeof(pReqBody->AppID) > GlobalDataManager::GetInstance().getAppID().length()
         ? GlobalDataManager::GetInstance().getAppID().length() : sizeof(pReqBody->AppID));
  memcpy(pReqBody->ClientIp,
         GlobalDataManager::GetInstance().getClientIp().c_str(),
         sizeof(pReqBody->ClientIp) > GlobalDataManager::GetInstance().getClientIp().length()
         ? GlobalDataManager::GetInstance().getClientIp().length() : sizeof(pReqBody->ClientIp));
  memcpy(pReqBody->ClientMac,
         GlobalDataManager::GetInstance().getClientMac().c_str(),
         sizeof(pReqBody->ClientMac) > GlobalDataManager::GetInstance().getClientMac().length()
         ? GlobalDataManager::GetInstance().getClientMac().length() : sizeof(pReqBody->ClientMac));
  pReqBody->Operway = GlobalDataManager::GetInstance().getOperWay()[0];
  pReqBody->Market = GlobalDataManager::GetInstance().getMarket();
  pReqBody->SubClientIndex = GlobalDataManager::GetInstance().getNode();
  memcpy(pReqBody->PcPrefix,
         GlobalDataManager::GetInstance().getPcPrefix().c_str(),
         sizeof(pReqBody->PcPrefix) > GlobalDataManager::GetInstance().getPcPrefix().length()
         ? GlobalDataManager::GetInstance().getPcPrefix().length() : sizeof(pReqBody->PcPrefix));
  memcpy(pReqBody->BusinessCode,
         GlobalDataManager::GetInstance().getBusinessCode().c_str(),
         sizeof(pReqBody->BusinessCode) > GlobalDataManager::GetInstance().getBusinessCode().length()
         ? GlobalDataManager::GetInstance().getBusinessCode().length() : sizeof(pReqBody->BusinessCode));
  memcpy(pReqBody->AppIDExt,
         GlobalDataManager::GetInstance().getAppIDExt().c_str(),
         sizeof(pReqBody->AppIDExt) > GlobalDataManager::GetInstance().getAppIDExt().length()
         ? GlobalDataManager::GetInstance().getAppIDExt().length() : sizeof(pReqBody->AppIDExt));

  /// Set EPOLL Out Event
  createSendEvent(MID_REQ_USER_LOGIN, MessageLength, TcpType::ManagerTcp);
}

bool ClientDemo::isManagerLogin() const {
  return m_managerLogin;
}

bool ClientDemo::isCounterQueryLogin() const {
  return m_counterQueryLogin;
}

bool ClientDemo::isCounterTradeLogin() const {
  return m_counterTradeLogin;
}

void ClientDemo::TestCounterLogin() {
  memset(m_sendBufCounterQuery, 0, sizeof(m_sendBufCounterQuery));

  auto MessageLength = sizeof(CXeleReqUserLoginField) + sizeof(CXeleFairReqHeadField)
      + sizeof(SysLocalInfoField) + sizeof(ExternInfoField)
      + sizeof(AllSuperviseInfoField);
  createFairHead(m_sendBufCounterQuery, TcpType::CounterQueryTcp, TID_REQ_USER_LOGIN, MessageLength);

  auto pReqBody = (CXeleReqUserLoginField*) (m_sendBufCounterQuery + sizeof(CXeleFairReqHeadField));
  memcpy(pReqBody->AccountID,
         GlobalDataManager::GetInstance().getAccountInfo().accountId.c_str(),
         sizeof(pReqBody->AccountID) > GlobalDataManager::GetInstance().getAccountInfo().accountId.length()
         ? GlobalDataManager::GetInstance().getAccountInfo().accountId.length() : sizeof(pReqBody->AccountID));
  memcpy(pReqBody->Password,
         GlobalDataManager::GetInstance().getAccountInfo().password.c_str(),
         sizeof(pReqBody->Password) > GlobalDataManager::GetInstance().getAccountInfo().password.length()
         ? GlobalDataManager::GetInstance().getAccountInfo().password.length() : sizeof(pReqBody->Password));

  memcpy(pReqBody->AppID,
         GlobalDataManager::GetInstance().getAppID().c_str(),
         sizeof(pReqBody->AppID) > GlobalDataManager::GetInstance().getAppID().length()
         ? GlobalDataManager::GetInstance().getAppID().length() : sizeof(pReqBody->AppID));
  memcpy(pReqBody->ClientIp,
         GlobalDataManager::GetInstance().getClientIp().c_str(),
         sizeof(pReqBody->ClientIp) > GlobalDataManager::GetInstance().getClientIp().length()
         ? GlobalDataManager::GetInstance().getClientIp().length() : sizeof(pReqBody->ClientIp));
  memcpy(pReqBody->ClientMac,
         GlobalDataManager::GetInstance().getClientMac().c_str(),
         sizeof(pReqBody->ClientMac) > GlobalDataManager::GetInstance().getClientMac().length()
         ? GlobalDataManager::GetInstance().getClientMac().length() : sizeof(pReqBody->ClientMac));
  pReqBody->SubClientIndex = GlobalDataManager::GetInstance().getNode();
  pReqBody->Operway = GlobalDataManager::GetInstance().getOperWay()[0];
  pReqBody->ProtocolFlag = API_PROTOCOL_3_0;
  pReqBody->SoftRspRcv = 0;
  memcpy(pReqBody->PcPrefix,
         GlobalDataManager::GetInstance().getPcPrefix().c_str(),
         sizeof(pReqBody->PcPrefix) > GlobalDataManager::GetInstance().getPcPrefix().length()
         ? GlobalDataManager::GetInstance().getPcPrefix().length() : sizeof(pReqBody->PcPrefix));
  memcpy(pReqBody->HdPartitionInfo,
         GlobalDataManager::GetInstance().getHdPartitionInfo().c_str(),
         sizeof(pReqBody->HdPartitionInfo) > GlobalDataManager::GetInstance().getHdPartitionInfo().length()
         ? GlobalDataManager::GetInstance().getHdPartitionInfo().length() : sizeof(pReqBody->HdPartitionInfo));
  memcpy(pReqBody->BusinessCode,
         GlobalDataManager::GetInstance().getBusinessCode().c_str(),
         sizeof(pReqBody->BusinessCode) > GlobalDataManager::GetInstance().getBusinessCode().length()
         ? GlobalDataManager::GetInstance().getBusinessCode().length() : sizeof(pReqBody->BusinessCode));
  memcpy(pReqBody->AppIDExt,
         GlobalDataManager::GetInstance().getAppIDExt().c_str(),
         sizeof(pReqBody->AppIDExt) > GlobalDataManager::GetInstance().getAppIDExt().length()
         ? GlobalDataManager::GetInstance().getAppIDExt().length() : sizeof(pReqBody->AppIDExt));
  pReqBody->FpgaRspRcv = 0;

  /// 监管信息
  auto allInfo = (AllSuperviseInfoField*)(m_sendBufCounterQuery + sizeof(CXeleReqUserLoginField) + sizeof(CXeleFairReqHeadField)
      + sizeof(SysLocalInfoField) + sizeof(ExternInfoField));
  memcpy(allInfo->AllSuperviseInfo,
         GlobalDataManager::GetInstance().getAllSuperviseInfo().c_str(),
         sizeof(allInfo->AllSuperviseInfo) > GlobalDataManager::GetInstance().getAllSuperviseInfo().length()
         ? GlobalDataManager::GetInstance().getAllSuperviseInfo().length() : sizeof(allInfo->AllSuperviseInfo));

  /// Set EPOLL Out Event
  createSendEvent(TID_REQ_USER_LOGIN, MessageLength, TcpType::CounterQueryTcp);
}

void ClientDemo::TestInitTrade() {
  /// Set EPOLL Out Event
  if (GlobalDataManager::GetInstance().isIsConnectHardTrade()) {
    memset(m_sendBufCounterQuery, 0, sizeof(m_sendBufCounterQuery));

    auto MessageLength = sizeof(CXeleFairReqHeadField) + sizeof(CXeleReqInitTraderField);
    createFairHead(m_sendBufCounterQuery, TcpType::CounterQueryTcp, TID_REQ_INIT_TRADER, MessageLength);

    auto pReqBody = (CXeleReqInitTraderField*) (m_sendBufCounterQuery + sizeof(CXeleFairReqHeadField));
    memcpy(pReqBody->AccountID,
           GlobalDataManager::GetInstance().getAccountInfo().accountId.c_str(),
           sizeof(pReqBody->AccountID) > GlobalDataManager::GetInstance().getAccountInfo().accountId.length()
           ? GlobalDataManager::GetInstance().getAccountInfo().accountId.length() : sizeof(pReqBody->AccountID));
    memcpy(pReqBody->TradeDestIp,
           GlobalDataManager::GetInstance().getCounterTradeIp().c_str(),
           sizeof(pReqBody->TradeDestIp) > GlobalDataManager::GetInstance().getCounterTradeIp().length()
           ? GlobalDataManager::GetInstance().getCounterTradeIp().length() : sizeof(pReqBody->TradeDestIp));
    pReqBody->TradeDestPort = GlobalDataManager::GetInstance().getCounterTradePort();
    /// todo:
    memcpy(pReqBody->TradeSrcIp,
           GlobalDataManager::GetInstance().getTradeSourceIp().c_str(),
           sizeof(pReqBody->TradeSrcIp) > GlobalDataManager::GetInstance().getTradeSourceIp().length()
           ? GlobalDataManager::GetInstance().getTradeSourceIp().length() : sizeof(pReqBody->TradeSrcIp));
    pReqBody->TradeSrcPort = GlobalDataManager::GetInstance().getTradeSoucrePort();

    /// 发往柜台的查询链路 TcpType::CounterQueryTcp
    PRINT_INFO("send to query tcp");
    createSendEvent(TID_REQ_INIT_TRADER, MessageLength, TcpType::CounterQueryTcp);
  } else {
    memset(m_sendBufCounterTrade, 0, sizeof(m_sendBufCounterTrade));

    auto MessageLength = sizeof(CXeleFairReqHeadField) + sizeof(CXeleReqInitTraderField);
    createFairHead(m_sendBufCounterTrade, TcpType::CounterQueryTcp, TID_REQ_INIT_TRADER, MessageLength);

    auto pReqBody = (CXeleReqInitTraderField*) (m_sendBufCounterTrade + sizeof(CXeleFairReqHeadField));
    memcpy(pReqBody->AccountID,
           GlobalDataManager::GetInstance().getAccountInfo().accountId.c_str(),
           sizeof(pReqBody->AccountID) > GlobalDataManager::GetInstance().getAccountInfo().accountId.length()
           ? GlobalDataManager::GetInstance().getAccountInfo().accountId.length() : sizeof(pReqBody->AccountID));
    memcpy(pReqBody->TradeDestIp,
           GlobalDataManager::GetInstance().getCounterTradeIp().c_str(),
           sizeof(pReqBody->TradeDestIp) > GlobalDataManager::GetInstance().getCounterTradeIp().length()
           ? GlobalDataManager::GetInstance().getCounterTradeIp().length() : sizeof(pReqBody->TradeDestIp));
    pReqBody->TradeDestPort = GlobalDataManager::GetInstance().getCounterTradePort();
    /// todo:
    memcpy(pReqBody->TradeSrcIp,
           GlobalDataManager::GetInstance().getTradeSourceIp().c_str(),
           sizeof(pReqBody->TradeSrcIp) > GlobalDataManager::GetInstance().getTradeSourceIp().length()
           ? GlobalDataManager::GetInstance().getTradeSourceIp().length() : sizeof(pReqBody->TradeSrcIp));
    pReqBody->TradeSrcPort = GlobalDataManager::GetInstance().getTradeSoucrePort();

    /// 发往柜台的交易链路 TcpType::CounterTradeTcp
    PRINT_INFO("send to trade tcp");
    createSendEvent(TID_REQ_INIT_TRADER, MessageLength, TcpType::CounterTradeTcp);
  }
}

/// 测试裸协议报单
void ClientDemo::TestOrderInsert() {
  memset(m_sendBufCounterTrade, 0, sizeof(m_sendBufCounterTrade));

  auto MessageLength = sizeof(CXeleFairReqHeadField) + sizeof(CXeleReqOrderInsertField);
  createFairHead(m_sendBufCounterTrade, TcpType::CounterTradeTcp, TID_REQ_ORDER_INSERT, MessageLength);

  /// Construct ReqBody
  auto pReqInsert = (CXeleReqOrderInsertField*)(m_sendBufCounterTrade + sizeof(CXeleFairReqHeadField));

  pReqInsert->UserLocalID = GlobalDataManager::GetInstance().maxLocalId();
  memcpy(pReqInsert->SecuritiesID,
         GlobalDataManager::GetInstance().getSecuritiesID().c_str(),
         sizeof(pReqInsert->SecuritiesID) > GlobalDataManager::GetInstance().getSecuritiesID().length()
         ? GlobalDataManager::GetInstance().getSecuritiesID().length() : sizeof(pReqInsert->SecuritiesID));
  pReqInsert->LimitPrice = GlobalDataManager::GetInstance().getLimitPrice();
  pReqInsert->Volume = GlobalDataManager::GetInstance().getVolume();
  pReqInsert->Direction = GlobalDataManager::GetInstance().getDirection();
  pReqInsert->Operway = GlobalDataManager::GetInstance().getOperWay()[0];
  pReqInsert->OrderType = XELE_LIMIT_PRICE_TYPE;
  pReqInsert->TimeCondition = XELE_TIMEINFORCE_TYPE_GFD;

  /// Set EPOLL Out Event
  createSendEvent(TID_REQ_ORDER_INSERT, MessageLength, TcpType::CounterTradeTcp);
}

/// 测试裸协议撤单
void ClientDemo::TestOrderAction() {
  if (GlobalDataManager::GetInstance().orderTableEmpty()) {
    PRINT_INFO("OrderTable Is Empty, Skip OrderAction.");
    return;
  }
  memset(m_sendBufCounterTrade, 0, sizeof(m_sendBufCounterTrade));
  auto MessageLength = sizeof(CXeleFairReqHeadField) + sizeof(CXeleReqOrderActionField);
  createFairHead(m_sendBufCounterTrade, TcpType::CounterTradeTcp, TID_REQ_ORDER_ACTION, MessageLength);

  /// Construct ReqBody
  auto pReqAction = (CXeleReqOrderActionField*)(m_sendBufCounterTrade+sizeof(CXeleFairReqHeadField));
  pReqAction->UserLocalID = GlobalDataManager::GetInstance().maxLocalId();
  uint32_t originId = GlobalDataManager::GetInstance().getOrderTable()[0].orderLocalId;
  pReqAction->OrigSysID = originId;
  pReqAction->Operway = GlobalDataManager::GetInstance().getOperWay()[0];

  /// Set EPOLL Out Event
  createSendEvent(TID_REQ_ORDER_ACTION, MessageLength, TcpType::CounterTradeTcp);
}

void ClientDemo::TestOrderQuery() {
  memset(m_sendBufCounterQuery, 0, sizeof(m_sendBufCounterQuery));
  auto MessageLength = sizeof(CXeleFairReqHeadField) + sizeof(CXeleReqQryOrderField);
  createFairHead(m_sendBufCounterQuery, TcpType::CounterQueryTcp, TID_REQ_QRY_ORDER, MessageLength);

  /// Construct ReqBody
  auto pReqOrderQuery = (CXeleReqQryOrderField*)(m_sendBufCounterQuery + sizeof(CXeleFairReqHeadField));
  memcpy(pReqOrderQuery->SecuritiesID,
         GlobalDataManager::GetInstance().getSecuritiesID().c_str(),
         sizeof(pReqOrderQuery->SecuritiesID) > GlobalDataManager::GetInstance().getSecuritiesID().length()
         ? GlobalDataManager::GetInstance().getSecuritiesID().length() : sizeof(pReqOrderQuery->SecuritiesID));
  /// Set EPOLL Out Event
  createSendEvent(TID_REQ_QRY_ORDER, MessageLength, TcpType::CounterQueryTcp);
}

void ClientDemo::TestCounterLogout() {
  memset(m_sendBufCounterQuery, 0, sizeof(m_sendBufCounterQuery));
  auto MessageLength = sizeof(CXeleFairReqHeadField) + sizeof(CXeleReqUserLogoutField);
  createFairHead(m_sendBufCounterQuery, TcpType::CounterQueryTcp, TID_REQ_USER_LOGOUT, MessageLength);

  /// Construct ReqBody
  auto pReqUserLogout = (CXeleReqUserLogoutField*)(m_sendBufCounterQuery + sizeof(CXeleFairReqHeadField));
  memcpy(pReqUserLogout->AccountID,
         GlobalDataManager::GetInstance().getAccountInfo().accountId.c_str(),
         sizeof(pReqUserLogout->AccountID) > GlobalDataManager::GetInstance().getAccountInfo().accountId.length()
         ? GlobalDataManager::GetInstance().getAccountInfo().accountId.length() : sizeof(pReqUserLogout->AccountID));
  /// Set EPOLL Out Event
  createSendEvent(TID_REQ_USER_LOGOUT, MessageLength, TcpType::CounterQueryTcp);
}

void ClientDemo::TestManagerLogout() {
  memset(m_sendBufManager, 0, sizeof(m_sendBufManager));
  auto MessageLength = sizeof(CXeleFairReqHeadField) + sizeof(CXeleReqUserLogoutManagerField);
  createFairHead(m_sendBufManager, TcpType::ManagerTcp, MID_REQ_USER_LOGOUT, MessageLength);

  /// Construct ReqBody
  auto pReqUserLogout = (CXeleReqUserLogoutManagerField*)(m_sendBufManager + sizeof(CXeleFairReqHeadField));
  memcpy(pReqUserLogout->AccountID,
         GlobalDataManager::GetInstance().getAccountInfo().accountId.c_str(),
         sizeof(pReqUserLogout->AccountID) > GlobalDataManager::GetInstance().getAccountInfo().accountId.length()
         ? GlobalDataManager::GetInstance().getAccountInfo().accountId.length() : sizeof(pReqUserLogout->AccountID));
  pReqUserLogout->Market = GlobalDataManager::GetInstance().getMarket();
  /// Set EPOLL Out Event
  createSendEvent(MID_REQ_USER_LOGOUT, MessageLength, TcpType::ManagerTcp);
}