#include "TradeClient.h"
#include "Msg.h"
#include "message_generated.h"
#include "Utils/PackMsg.h"
#include "Utils/ZmqHelper.hpp"
#include "Utils/Logger.h"
#include "Utils/Config.h"
#include "Utils/TimeUtils.hpp"
#include
#include
#include
#include
#include
#include

TradeClient::TradeClient()
:m_pSendTradeReqContext(nullptr)
,m_pSendTradeReqSocket(nullptr)
,m_pRecvTradeRtnContext(nullptr)
,m_pRecvTradeRtnSocket(nullptr)
,m_pSendQueryReqContext(nullptr)
,m_pSendQueryReqSocket(nullptr)
,m_pRecvQueryRtnContext(nullptr)
,m_pRecvQueryRtnSocket(nullptr)
{
m_tcp_keep_alive = 1;
m_tcp_keep_idle = 30;
m_tcp_keep_intvl = 5;
m_tcp_keep_cnt = 5;
m_router_handover = 1;
m_heartbeat_ivl = 1000;
m_heartbeat_timeout = 3000;
m_heartbeat_ttl = 6000;
m_last_heartbeat.store(-1);
m_gateway_is_live.store(false);
}

TradeClient::~TradeClient()
{

}

void TradeClient::handleLoginRsp(FbMsg::LoginRspT *loginRspT)
{
Logger::getInstance().log->debug("TradeClient::handleLoginRsp");
LoginRsp loginRsp;
loginRsp.AccountId = loginRspT->AccountId;
loginRsp.ChannelId = loginRspT->ChannelId;
loginRsp.LoginTime = loginRspT->LoginTime;
loginRsp.TradingDay = loginRspT->TradingDay;
loginRsp.MaxLocalId = loginRspT->MaxLocalId;
}

void TradeClient::handleEntrustRtn(FbMsg::OrderT *orderT)
{
Logger::getInstance().log->info("TradeClient::handleEntrustRtn ChannelId:{} ExchangeId:{} InstrumentId:{} "
"AccountId:{} Volume:{} LimitPrice:{} BuyOrSell:{} OrderPriceType:{} OrderStatus:{} OrderId:{} OrderSysId:{} VolumeLeft:{} VolumeTraded:{} InsertTime:{} StatusMsg:{}"
,orderT->ChannelId,orderT->ExchangeId,orderT->InstrumentId,orderT->AccountId,orderT->Volume
,orderT->LimitPrice,orderT->BuyOrSell,orderT->OrderPriceType,orderT->OrderStatus,orderT->OrderId,orderT->OrderSysId,orderT->VolumeLeft,orderT->VolumeTraded,orderT->InsertTime,orderT->StatusMsg);
Order order;
order.ChannelId = orderT->ChannelId;
order.ExchangeId = orderT->ExchangeId;
order.InstrumentId = orderT->InstrumentId;
order.AccountId = orderT->AccountId;
order.OrderSysId = orderT->OrderSysId;
order.Volume = orderT->Volume;
order.VolumeLeft = orderT->VolumeLeft;
order.VolumeTraded = orderT->VolumeTraded;
order.LimitPrice = orderT->LimitPrice;
order.TradeAmount = orderT->TradeAmount;
order.BuyOrSell = orderT->BuyOrSell;
order.HedgeFlag = orderT->HedgeFlag;
order.OrderPriceType = orderT->OrderPriceType;
order.OpenOrClose = orderT->OpenOrClose;
order.OrderStatus = orderT->OrderStatus;
order.InsertTime = orderT->InsertTime;
order.LocalInsertTime = orderT->LocalInsertTime;
order.OrderId = orderT->OrderId;
order.OrigOrderSysId = orderT->OrigOrderSysId;
order.OrigOrderId = orderT->OrigOrderId;
order.StatusMsg = orderT->StatusMsg;

}

void TradeClient::handleCancelOrderRtn(FbMsg::OrderT *orderT)
{
Logger::getInstance().log->info("TradeClient::handleCancelOrderRtn ChannelId:{} AccountId:{} OrderId:{} OrderSysId:{} OrderStatus:{} StatusMsg:{}",orderT->ChannelId,orderT->AccountId,orderT->OrderId,orderT->OrderSysId,orderT->OrderStatus,orderT->StatusMsg);
Order order;
order.ChannelId = orderT->ChannelId;
order.ExchangeId = orderT->ExchangeId;
order.InstrumentId = orderT->InstrumentId;
order.AccountId = orderT->AccountId;
order.OrderSysId = orderT->OrderSysId;
order.Volume = orderT->Volume;
order.VolumeLeft = orderT->VolumeLeft;
order.VolumeTraded = orderT->VolumeTraded;
order.LimitPrice = orderT->LimitPrice;
order.TradeAmount = orderT->TradeAmount;
order.BuyOrSell = orderT->BuyOrSell;
order.HedgeFlag = orderT->HedgeFlag;
order.OrderPriceType = orderT->OrderPriceType;
order.OpenOrClose = orderT->OpenOrClose;
order.OrderStatus = orderT->OrderStatus;
order.InsertTime = orderT->InsertTime;
order.LocalInsertTime = orderT->LocalInsertTime;
order.OrderId = orderT->OrderId;
order.OrigOrderSysId = orderT->OrigOrderSysId;
order.OrigOrderId = orderT->OrigOrderId;
order.StatusMsg = orderT->StatusMsg;
}

void TradeClient::handleTradeRtn(FbMsg::TradeT *tradeT)
{
Logger::getInstance().log->info("TradeClient::handleTradeRtn ChannelId:{} ExchangeId:{} InstrumentId:{} "
"AccountId:{} OrderId:{} OrderSysId:{} TradeId:{} Price:{} Volume:{} TradeTime:{} BuyOrSell:{} "
,tradeT->ChannelId,tradeT->ExchangeId,tradeT->InstrumentId,tradeT->AccountId,tradeT->OrderId
,tradeT->OrderSysId,tradeT->TradeId,tradeT->Price,tradeT->Volume,tradeT->TradeTime,tradeT->BuyOrSell);
Trade trade;
trade.ChannelId = tradeT->ChannelId;
trade.ExchangeId = tradeT->ExchangeId;
trade.InstrumentId = tradeT->InstrumentId;
trade.AccountId = tradeT->AccountId;
trade.OrderSysId = tradeT->OrderSysId;
trade.OrderId = tradeT->OrderId;
trade.TradeId = tradeT->TradeId;
trade.Price = tradeT->Price;
trade.Volume = tradeT->Volume;
trade.TradeTime = TimeUtils::getCurrentYYYYMMDDHHMMSS();
trade.BuyOrSell = tradeT->BuyOrSell;
trade.HedgeFlag = tradeT->HedgeFlag;
trade.OpenOrClose = tradeT->OpenOrClose;

}

void TradeClient::handleQryFundAccountRsp(FbMsg::FundAccountT *fundAccountT)
{
Logger::getInstance().log->debug("TradeClient::handleQryFundAccountRsp ChannelId:{} AccountId:{}",fundAccountT->ChannelId,fundAccountT->AccountId);
FundAccount fundAccount;
fundAccount.ChannelId = fundAccountT->ChannelId;
fundAccount.AccountId = fundAccountT->AccountId;
fundAccount.Available = fundAccountT->Available;
fundAccount.Balance = fundAccountT->Balance;
}

void TradeClient::handleQryPositionsRsp(FbMsg::PositionT *positionT)
{
Logger::getInstance().log->debug("TradeClient::handleQryPositionsRsp ChannelId:{} AccountId:{}",positionT->ChannelId,positionT->AccountId);
Position position;
position.ChannelId = positionT->ChannelId;
position.AccountId = positionT->AccountId;
position.ExchangeId = positionT->ExchangeId;
position.InstrumentId = positionT->InstrumentId;
position.Name = positionT->Name;
position.Volume = positionT->Volume;
position.EnableVolume = positionT->EnableVolume;
position.YdVolume = positionT->YdVolume;
position.TdVolume = positionT->TdVolume;
position.AvgPrice = positionT->AvgPrice;
position.BuyOrSell = positionT->BuyOrSell;
position.HedgeFlag = positionT->HedgeFlag;
}

void TradeClient::handleQryOrdersRsp(FbMsg::OrderT *orderT)
{
Logger::getInstance().log->debug("TradeClient::handleQryOrdersRsp ChannelId:{} AccountId:{}",orderT->ChannelId,orderT->AccountId);
Order order;
order.ChannelId = orderT->ChannelId;
order.ExchangeId = orderT->ExchangeId;
order.InstrumentId = orderT->InstrumentId;
order.AccountId = orderT->AccountId;
order.OrderSysId = orderT->OrderSysId;
order.Volume = orderT->Volume;
order.VolumeLeft = orderT->VolumeLeft;
order.VolumeTraded = orderT->VolumeTraded;
order.LimitPrice = orderT->LimitPrice;
order.TradeAmount = orderT->TradeAmount;
order.BuyOrSell = orderT->BuyOrSell;
order.HedgeFlag = orderT->HedgeFlag;
order.OrderPriceType = orderT->OrderPriceType;
order.OpenOrClose = orderT->OpenOrClose;
order.OrderStatus = orderT->OrderStatus;
order.InsertTime = orderT->InsertTime;
order.LocalInsertTime = orderT->LocalInsertTime;
order.OrderId = orderT->OrderId;
order.OrigOrderSysId = orderT->OrigOrderSysId;
order.OrigOrderId = orderT->OrigOrderId;
order.StatusMsg = orderT->StatusMsg;
}

void TradeClient::handleQryTradesRsp(FbMsg::TradeT *tradeT)
{
Logger::getInstance().log->debug("TradeClient::handleQryTradesRsp ChannelId:{} AccountId:{}",tradeT->ChannelId,tradeT->AccountId);
Trade trade;
trade.ChannelId = tradeT->ChannelId;
trade.ExchangeId = tradeT->ExchangeId;
trade.InstrumentId = tradeT->InstrumentId;
trade.AccountId = tradeT->AccountId;
trade.OrderSysId = tradeT->OrderSysId;
trade.OrderId = tradeT->OrderId;
trade.TradeId = tradeT->TradeId;
trade.Price = tradeT->Price;
trade.Volume = tradeT->Volume;
trade.TradeTime = tradeT->TradeTime;
trade.BuyOrSell = tradeT->BuyOrSell;
trade.HedgeFlag = tradeT->HedgeFlag;
trade.OpenOrClose = tradeT->OpenOrClose;
}

void TradeClient::handleQryShareholderRsp(FbMsg::ShareHolderT *shareHolderT)
{
Logger::getInstance().log->debug("TradeClient::handleQryShareholderRsp ChannelId:{} AccountId:{}",shareHolderT->ChannelId,shareHolderT->AccountId);
ShareHolder shareHolder;
shareHolder.ChannelId = shareHolderT->ChannelId;
shareHolder.AccountId = shareHolderT->AccountId;
shareHolder.ExchangeId = shareHolderT->ExchangeId;
shareHolder.ShareHolderId = shareHolderT->ShareHolderId;
shareHolder.ShareHolderName = shareHolderT->ShareHolderName;
}

void TradeClient::handleError(FbMsg::ErrorInfoT *errorInfoT)
{
Logger::getInstance().log->error("TradeClient::handleError ChannelId:{} AccountId:{} ErrorId:{} ErrorMsg:{}",errorInfoT->ChannelId,errorInfoT->AccountId,errorInfoT->ErrorId,errorInfoT->ErrorMsg);
ErrorInfo error;
error.ChannelId = errorInfoT->ChannelId;
error.AccountId = errorInfoT->AccountId;
error.ErrorId = errorInfoT->ErrorId;
error.ErrorMsg = errorInfoT->ErrorMsg;
}

void TradeClient::handleHeartBeatRsp(FbMsg::HeartBeatT *heartT)
{
Logger::getInstance().log->debug("TradeClient::handleHeartBeatRsp ChannelId:{} AccountId:{}",heartT->ChannelId,heartT->AccountId);
HeartBeat heart;
heart.ChannelId = heartT->ChannelId;
heart.AccountId = heartT->AccountId;
heart.LastTime = heartT->LastTime;
heart.ErrorId = heartT->ErrorId;
heart.ErrorMsg = heartT->ErrorMsg;
if(!m_gateway_is_live.load())
{
m_gateway_is_live.store(true);
}
m_last_heartbeat.store(std::stoll(TimeUtils::getCurrentTimestamp()));
}

void TradeClient::handleAgentRegister(FbMsg::AgentInfoT *agentT)
{
// Logger::getInstance().log->debug("TradeClient::handleAgentRegister ChannelId:{} AccountId:{}" ,agentT->ChannelId, agentT->AccountId);
// std::string agentid = agentT->ChannelId+agentT->AccountId;
// AgentInfo agent;
// agent.ChannelId = agentT->ChannelId;
// agent.AccountId = agentT->AccountId;
// agent.LastTime = TimeUtils::getCurrentTimestamp();
// agent.Status = agentT->Status;
// {
// std::lock_guardstd::mutex lock(m_mtxAgent);
// m_mapAgent[agentid] = agent;
// }
// _agentRegisterRsp(agent);
}

void TradeClient::run()
{
while(true)
{
LoginReq loginReq;
loginReq.ChannelId = m_channelid;
loginReq.AccountId = m_accountid;
loginReq.Password = m_pass;
login(loginReq);
std::this_thread::sleep_for(std::chrono::seconds(1));


    Order insOrder;
    insOrder.ChannelId = m_channelid;
    insOrder.AccountId = m_accountid;
    insOrder.ExchangeId = EXCHANGE_SH;
    insOrder.InstrumentId = "600018";
    insOrder.LimitPrice = 76;
    insOrder.Volume = 200;
    insOrder.BuyOrSell = ORDER_DIRECTION_BUY;
    insOrder.OrderPriceType = ORDER_PRICE_TYPE_LIMIT;
    insOrder.OrderId = "1000000";
    insOrder.ChannelId = m_channelid;
    insOrder.AccountId = m_accountid;
    insertOrder(insOrder);
    std::this_thread::sleep_for(std::chrono::seconds(1));
    
    Order canOrder;
    canOrder.ChannelId = m_channelid;
    canOrder.AccountId = m_accountid;
    canOrder.OrderId = "1000001";
    canOrder.OrderSysId="1000000";
    cancelOrder(canOrder);
    std::this_thread::sleep_for(std::chrono::seconds(1));
    
    QryReq query;
    query.ChannelId = m_channelid;
    query.AccountId = m_accountid;
    queryAccount(query);
    std::this_thread::sleep_for(std::chrono::seconds(1));
    
    queryPositions(query);
    std::this_thread::sleep_for(std::chrono::seconds(1));
    
    queryOrders(query);
    std::this_thread::sleep_for(std::chrono::seconds(1));
    
    queryTrades(query);
    std::this_thread::sleep_for(std::chrono::seconds(1));
    
    queryShareHolderInfo(query);
    std::this_thread::sleep_for(std::chrono::seconds(1));
}
}

void TradeClient::init(ConfigParams &params)
{
Logger::getInstance().log->debug("{}",GET_CLASS_FUNCTION_NAME);
m_send_trade_req_addr = params["client"]["send_trade_req_addr"].asstd::string();
m_recv_trade_rtn_addr = params["client"]["recv_trade_rtn_addr"].asstd::string();
m_send_query_req_addr = params["client"]["send_query_req_addr"].asstd::string();
m_recv_query_rtn_addr = params["client"]["recv_query_rtn_addr"].asstd::string();
m_heartBeatInterval = params["client"]["heart_beat_interval"].as();
m_checkHeartBeatTimeout = params["client"]["check_heart_beat_timeout"].as();
m_channelid = params["client"]["channelid"].asstd::string();
m_accountid = params["client"]["accountid"].asstd::string();
m_pass = params["client"]["pass"].asstd::string();


Logger::getInstance().log->debug("{} load config recv_trade_req_addr:{} send_trade_rtn_addr:{} recv_query_req_addr:{} send_query_rtn_addr:{}",GET_CLASS_FUNCTION_NAME,m_send_trade_req_addr,m_recv_trade_rtn_addr,m_send_query_req_addr,m_recv_query_rtn_addr);
_createZmqCtx();
initTrade();
initQuery();
std::thread sendHeartBeat(&TradeClient::_sendHeartBeat, this);
sendHeartBeat.detach();
std::thread checkTimeout(&TradeClient::_checkHeartBeatTimeout, this);
checkTimeout.detach();
}

void TradeClient::initTrade()
{
Logger::getInstance().log->debug("{}",GET_CLASS_FUNCTION_NAME);
m_send_trade_req_identity = m_channelid+m_accountid;
m_recv_trade_rtn_identity = m_channelid+m_accountid;
_setDealerSocket(m_pSendTradeReqSocket, m_send_trade_req_identity, m_send_trade_req_addr);
_setDealerSocket(m_pRecvTradeRtnSocket, m_recv_trade_rtn_identity, m_recv_trade_rtn_addr);
std::thread recvTraderRtn(&TradeClient::listenTradeRtnMessages, this);
recvTraderRtn.detach();
}

void TradeClient::initQuery()
{
Logger::getInstance().log->debug("{}",GET_CLASS_FUNCTION_NAME);
m_send_query_req_identity = m_channelid+m_accountid;
m_recv_query_rtn_identity = m_channelid+m_accountid;
_setDealerSocket(m_pSendQueryReqSocket, m_send_query_req_identity, m_send_query_req_addr);
_setDealerSocket(m_pRecvQueryRtnSocket, m_recv_query_rtn_identity, m_recv_query_rtn_addr);
std::thread recvQueryRtn(&TradeClient::listenQueryRtnMessages, this);
recvQueryRtn.detach();
}

void TradeClient::listenTradeRtnMessages()
{
Logger::getInstance().log->debug("{} start",GET_CLASS_FUNCTION_NAME);
const int BUFFSIZE = 1024;
uchar_t buff[BUFFSIZE];


while (1) 
{
    zmq_pollitem_t items[] = {{ m_pRecvTradeRtnSocket, 0, ZMQ_POLLIN, 0}};
    int rc = zmq_poll(items, 1, -1);
    if (rc == -1) {
        Logger::getInstance().log->error("{} Poll error:{}",GET_CLASS_FUNCTION_NAME,zmq_strerror(errno));
        continue;
    }
    
    if (items[0].revents & ZMQ_POLLIN)
    {
        memset(buff,0,sizeof(buff));
        ZmqHelper::recvBuffMsg(m_pRecvTradeRtnSocket,buff);
        
        Header header;
        memcpy(&header,buff,sizeof(Header));
        if(header.MsgType==MsgType::LoginRspType)
        {
            auto loginRsp = flatbuffers::GetRoot<FbMsg::LoginRsp>(buff+sizeof(Header));
            FbMsg::LoginRspT* loginRspT = loginRsp->UnPack();
            handleLoginRsp(loginRspT);
        }
        else if(header.MsgType==MsgType::EntrustRtnType)
        {
            auto order = flatbuffers::GetRoot<FbMsg::Order>(buff+sizeof(Header));
            FbMsg::OrderT* orderT = order->UnPack();
            handleEntrustRtn(orderT);
        }
        else if(header.MsgType==MsgType::CancelOrderRtnType)
        {
            auto order = flatbuffers::GetRoot<FbMsg::Order>(buff+sizeof(Header));
            FbMsg::OrderT* orderT = order->UnPack();
            handleCancelOrderRtn(orderT);
        }
        else if(header.MsgType==MsgType::TradeRtnType)
        {
            auto trade = flatbuffers::GetRoot<FbMsg::Trade>(buff+sizeof(Header));
            FbMsg::TradeT* tradeT = trade->UnPack();
            handleTradeRtn(tradeT);
        }
        else if(header.MsgType==MsgType::ErrorInfoType)
        {
            auto error = flatbuffers::GetRoot<FbMsg::ErrorInfo>(buff+sizeof(Header));
            FbMsg::ErrorInfoT* errorT = error->UnPack();
            handleError(errorT);
        }
        else if(header.MsgType==MsgType::HeartBeatTypeRspType)
        {
            auto heart = flatbuffers::GetRoot<FbMsg::HeartBeat>(buff+sizeof(Header));
            FbMsg::HeartBeatT* heartT = heart->UnPack();
            handleHeartBeatRsp(heartT);
        }
        else if(header.MsgType==MsgType::AgentRegisterType)
        {
            auto agent = flatbuffers::GetRoot<FbMsg::AgentInfo>(buff+sizeof(Header));
            FbMsg::AgentInfoT* agentT = agent->UnPack();
            handleAgentRegister(agentT);
        }
        memset(buff,0,sizeof(buff));
    }
}
zmq_close(m_pRecvTradeRtnSocket);
zmq_ctx_destroy(m_pRecvTradeRtnContext);
return;
}

void TradeClient::listenQueryRtnMessages()
{
Logger::getInstance().log->debug("{} start",GET_CLASS_FUNCTION_NAME);
const int BUFFSIZE = 1024;
uchar_t buff[BUFFSIZE];


while (1) 
{
    zmq_pollitem_t items[] = {{ m_pRecvQueryRtnSocket, 0, ZMQ_POLLIN, 0}};
    int rc = zmq_poll(items, 1, -1);
    if (rc == -1) {
        Logger::getInstance().log->error("{} Poll error:{}",GET_CLASS_FUNCTION_NAME,zmq_strerror(errno));
        continue;
    }
    
    if (items[0].revents & ZMQ_POLLIN)
    {
        int ret = 0;
        memset(buff,0,sizeof(buff));
        ZmqHelper::recvBuffMsg(m_pRecvQueryRtnSocket,buff);
        
        Header header;
        memcpy(&header,buff,sizeof(Header));
        if(header.MsgType==MsgType::QryFundAccRspType)
        {
            auto fundAccount = flatbuffers::GetRoot<FbMsg::FundAccount>(buff+sizeof(Header));
            FbMsg::FundAccountT* fundAccountT = fundAccount->UnPack();
            handleQryFundAccountRsp(fundAccountT);
        }
        else if(header.MsgType==MsgType::QryPositionsRspType)
        {
            auto position = flatbuffers::GetRoot<FbMsg::Position>(buff+sizeof(Header));
            FbMsg::PositionT* positionT = position->UnPack();
            handleQryPositionsRsp(positionT);
        }
        else if(header.MsgType==MsgType::QryOrdersRspType)
        {
            auto order = flatbuffers::GetRoot<FbMsg::Order>(buff+sizeof(Header));
            FbMsg::OrderT* orderT = order->UnPack();
            handleQryOrdersRsp(orderT);
        }
        else if(header.MsgType==MsgType::QryTradesRspType)
        {
            auto trade = flatbuffers::GetRoot<FbMsg::Trade>(buff+sizeof(Header));
            FbMsg::TradeT* tradeT = trade->UnPack();
            handleQryTradesRsp(tradeT);
        }
        else if(header.MsgType==MsgType::QryShareHolderRspType)
        {
            auto shareHolder = flatbuffers::GetRoot<FbMsg::ShareHolder>(buff+sizeof(Header));
            FbMsg::ShareHolderT* shareHolderT = shareHolder->UnPack();
            handleQryShareholderRsp(shareHolderT);
        }
        else if(header.MsgType==MsgType::ErrorInfoType)
        {
            auto error = flatbuffers::GetRoot<FbMsg::ErrorInfo>(buff+sizeof(Header));
            FbMsg::ErrorInfoT* errorT = error->UnPack();
            handleError(errorT);
        }
        memset(buff,0,sizeof(buff)); 
    }
}
zmq_close(m_pRecvQueryRtnSocket);
zmq_ctx_destroy(m_pRecvQueryRtnContext);
return;
}

int TradeClient::login(LoginReq &loginReq)
{
if(!_checkGatewayStatus(loginReq.ChannelId+loginReq.AccountId)) // agent不在线
{
ErrorInfo error;
error.ChannelId = loginReq.ChannelId;
error.AccountId = loginReq.AccountId;
error.ErrorId = ERROR_ID_LOGIN;
error.ErrorMsg = "login gateway error";


    LoginRsp loginRsp;
    loginRsp.AccountId = loginReq.AccountId;
    loginRsp.ChannelId = loginReq.ChannelId;
    loginRsp.LoginTime = TimeUtils::getHH_MM_SSTime();
    loginRsp.TradingDay = TimeUtils::getDateYYYYMMDD();
    loginRsp.MaxLocalId = 10000000;
    return 0;
}
FbMsg::LoginReqT loginReqT;
loginReqT.AccountId = loginReq.AccountId;
loginReqT.Password = loginReq.Password;
loginReqT.ChannelId = loginReq.ChannelId;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbLoginReq = FbMsg::CreateLoginReq(builder,&loginReqT);
builder.Finish(fbLoginReq);

Header header;
header.MsgLength = sizeof(Header) + builder.GetSize();
header.MsgType = MsgType::LoginReqType;
header.ChannelId = 101;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());

std::lock_guard<std::mutex> lock(m_trade_mtx);
ZmqHelper::sendPackMsg(m_pSendTradeReqSocket, &msg);
return 0;
}

int TradeClient::logout()
{
return 0;
}

int TradeClient::insertOrder(Order &order)
{


if(!_checkGatewayStatus(order.ChannelId+order.AccountId)) // agent不在线
{
    Logger::getInstance().log->error("TradeClient::insertOrder client not online, gateway insert order error ChannelId:{} ExchangeId:{} InstrumentId:{} "
                                "AccountId:{} Volume:{} LimitPrice:{} BuyOrSell:{} OrderPriceType:{} OrderId:{}"
                                ,order.ChannelId,order.ExchangeId,order.InstrumentId,order.AccountId,order.Volume
                                ,order.LimitPrice,order.BuyOrSell,order.OrderPriceType,order.OrderId);
    ErrorInfo error;
    error.ChannelId = order.ChannelId;
    error.AccountId = order.AccountId;
    error.ErrorId = ERROR_ID_INSERT;
    error.ErrorMsg = "TradeClient::insertOrder gateway insert order error|client not online";
    order.OrderStatus = ORDER_STATUS_REJECTED;
    order.VolumeTraded = 0;
    order.TradeAmount = 0;
    order.OrderSysId = std::string("#gateway-")+TimeUtils::getCurrentTimestamp();
    order.StatusMsg = error.ErrorMsg;
    return 0;
}
Logger::getInstance().log->info("TradeClient::insertOrder ChannelId:{} ExchangeId:{} InstrumentId:{} "
                                "AccountId:{} Volume:{} LimitPrice:{} BuyOrSell:{} OrderPriceType:{} OrderId:{}"
                                ,order.ChannelId,order.ExchangeId,order.InstrumentId,order.AccountId,order.Volume
                                ,order.LimitPrice,order.BuyOrSell,order.OrderPriceType,order.OrderId);
FbMsg::OrderT orderT;
orderT.ChannelId = order.ChannelId;
orderT.ExchangeId = order.ExchangeId;
orderT.InstrumentId = order.InstrumentId;
orderT.AccountId = order.AccountId;
orderT.OrderSysId = order.OrderSysId;
orderT.Volume = order.Volume;
orderT.VolumeLeft = order.VolumeLeft;
orderT.VolumeTraded = order.VolumeTraded;
orderT.LimitPrice = order.LimitPrice;
orderT.TradeAmount = order.TradeAmount;
orderT.BuyOrSell = order.BuyOrSell;
orderT.HedgeFlag = order.HedgeFlag;
orderT.OrderPriceType = order.OrderPriceType;
orderT.OpenOrClose = order.OpenOrClose;
orderT.OrderStatus = order.OrderStatus;
orderT.InsertTime = order.InsertTime;
orderT.LocalInsertTime = order.LocalInsertTime;
orderT.OrderId = order.OrderId;
orderT.OrigOrderSysId = order.OrigOrderSysId;
orderT.OrigOrderId = order.OrigOrderId;
orderT.StatusMsg = order.StatusMsg;

flatbuffers::FlatBufferBuilder builder(1024);
auto fbOrder = FbMsg::CreateOrder(builder,&orderT);
builder.Finish(fbOrder);

Header header;
header.MsgLength = sizeof(Header) + builder.GetSize();
header.MsgType = MsgType::EntrustType;
header.ChannelId = 101;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());

std::lock_guard<std::mutex> lock(m_trade_mtx);;
ZmqHelper::sendPackMsg(m_pSendTradeReqSocket, &msg);

return 0;
}

int TradeClient::cancelOrder(Order &order)
{
if(!_checkGatewayStatus(order.ChannelId+order.AccountId)) // agent不在线
{
Logger::getInstance().log->error("TradeClient::cancelOrder client not online, gateway cancel order error ChannelId:{} ExchangeId:{} InstrumentId:{} AccountId:{} OrderId:{} OrderSysId:{}"
,order.ChannelId,order.ExchangeId,order.InstrumentId,order.AccountId,order.OrderId,order.OrderSysId);
ErrorInfo error;
error.ChannelId = order.ChannelId;
error.AccountId = order.AccountId;
error.ErrorId = ERROR_ID_CANCEL;
error.ErrorMsg = "TradeClient::cancelOrder gateway cancel order error|client not online";
order.OrderStatus = ORDER_STATUS_REJECTED;
order.StatusMsg = error.ErrorMsg;
return 0;
}
Logger::getInstance().log->info("TradeClient::cancelOrder ChannelId:{} ExchangeId:{} InstrumentId:{} AccountId:{} OrderId:{} OrderSysId:{}"
,order.ChannelId,order.ExchangeId,order.InstrumentId,order.AccountId,order.OrderId,order.OrderSysId);
FbMsg::OrderT orderT;
orderT.ChannelId = order.ChannelId;
orderT.ExchangeId = order.ExchangeId;
orderT.InstrumentId = order.InstrumentId;
orderT.AccountId = order.AccountId;
orderT.OrderSysId = order.OrderSysId;
orderT.Volume = order.Volume;
orderT.VolumeLeft = order.VolumeLeft;
orderT.VolumeTraded = order.VolumeTraded;
orderT.LimitPrice = order.LimitPrice;
orderT.TradeAmount = order.TradeAmount;
orderT.BuyOrSell = order.BuyOrSell;
orderT.HedgeFlag = order.HedgeFlag;
orderT.OrderPriceType = order.OrderPriceType;
orderT.OpenOrClose = order.OpenOrClose;
orderT.OrderStatus = order.OrderStatus;
orderT.InsertTime = order.InsertTime;
orderT.LocalInsertTime = order.LocalInsertTime;
orderT.OrderId = order.OrderId;
orderT.OrigOrderSysId = order.OrigOrderSysId;
orderT.OrigOrderId = order.OrigOrderId;
orderT.StatusMsg = order.StatusMsg;


flatbuffers::FlatBufferBuilder builder(1024);
auto fbOrder = FbMsg::CreateOrder(builder,&orderT);
builder.Finish(fbOrder);

Header header;
header.MsgLength = sizeof(Header) + builder.GetSize();
header.MsgType = MsgType::CancelOrderType;
header.ChannelId = 101;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());

std::lock_guard<std::mutex> lock(m_trade_mtx);
ZmqHelper::sendPackMsg(m_pSendTradeReqSocket, &msg);

return 0;
}

int TradeClient::cancelBatchOrder(std::vector &vecOrders)
{
Logger::getInstance().log->info("TradeClient::cancelBatchOrder start");
for(auto& order : vecOrders)
{
cancelOrder(order);
}
Logger::getInstance().log->info("TradeClient::cancelBatchOrder end");
return 0;
}

int TradeClient::queryAccount(QryReq& query)
{
if(!_checkGatewayStatus(query.ChannelId+query.AccountId)) // agent不在线
{
Logger::getInstance().log->error("TradeClient::queryAccount client not online, gateway query account error ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
static int qryAccErrCnt = 0;
if(qryAccErrCnt > STACK_DEEP)
{
return 0;
}
ErrorInfo error;
error.ChannelId = query.ChannelId;
error.AccountId = query.AccountId;
error.ErrorId = ERROR_ID_QRYACC;
error.ErrorMsg = "gateway query account error";
qryAccErrCnt++;
qryAccErrCnt--;
return 0;
}
Logger::getInstance().log->info("TradeClient::queryAccount ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
FbMsg::QryReqT queryT;
queryT.ChannelId = query.ChannelId;
queryT.AccountId = query.AccountId;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbQryReq = FbMsg::CreateQryReq(builder,&queryT);
builder.Finish(fbQryReq);


Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::QryFundAccType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());

std::lock_guard<std::mutex> lock(m_qry_mtx);
ZmqHelper::sendPackMsg(m_pSendQueryReqSocket, &msg);

return 0;
}

int TradeClient::queryPositions(QryReq& query)
{
if(!_checkGatewayStatus(query.ChannelId+query.AccountId)) // agent不在线
{
Logger::getInstance().log->error("TradeClient::queryPositions client not online, gateway query positions error ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
static int qryPosErrCnt = 0;
if(qryPosErrCnt > STACK_DEEP)
{
return 0;
}
ErrorInfo error;
error.ChannelId = query.ChannelId;
error.AccountId = query.AccountId;
error.ErrorId = ERROR_ID_QRYPOS;
error.ErrorMsg = "gateway query positions error";
qryPosErrCnt++;
qryPosErrCnt--;
return 0;
}
Logger::getInstance().log->info("TradeClient::queryPositions ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
FbMsg::QryReqT queryT;
queryT.ChannelId = query.ChannelId;
queryT.AccountId = query.AccountId;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbQryReq = FbMsg::CreateQryReq(builder,&queryT);
builder.Finish(fbQryReq);


Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::QryPositionsType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());

std::lock_guard<std::mutex> lock(m_qry_mtx);
ZmqHelper::sendPackMsg(m_pSendQueryReqSocket, &msg);

return 0;
}

int TradeClient::queryOrders(QryReq& query)
{
if(!_checkGatewayStatus(query.ChannelId+query.AccountId)) // agent不在线
{
Logger::getInstance().log->error("TradeClient::queryOrders client not online, gateway query orders error ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
static int qryOrdErrCnt = 0;
if(qryOrdErrCnt > STACK_DEEP)
{
return 0;
}
ErrorInfo error;
error.ChannelId = query.ChannelId;
error.AccountId = query.AccountId;
error.ErrorId = ERROR_ID_QRYORDERS;
error.ErrorMsg = "gateway query orders error";
qryOrdErrCnt++;
qryOrdErrCnt--;
return 0;
}
Logger::getInstance().log->info("TradeClient::queryOrders ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
FbMsg::QryReqT queryT;
queryT.ChannelId = query.ChannelId;
queryT.AccountId = query.AccountId;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbQryReq = FbMsg::CreateQryReq(builder,&queryT);
builder.Finish(fbQryReq);


Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::QryOrdersType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());

std::lock_guard<std::mutex> lock(m_qry_mtx);
ZmqHelper::sendPackMsg(m_pSendQueryReqSocket, &msg);

return 0;
}

int TradeClient::queryTrades(QryReq& query)
{
if(!_checkGatewayStatus(query.ChannelId+query.AccountId)) // agent不在线
{
Logger::getInstance().log->error("TradeClient::queryTrades client not online, gateway query trades error ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
static int qryTrdErrCnt = 0;
if(qryTrdErrCnt > STACK_DEEP)
{
return 0;
}
ErrorInfo error;
error.ChannelId = query.ChannelId;
error.AccountId = query.AccountId;
error.ErrorId = ERROR_ID_QRYTRADES;
error.ErrorMsg = "gateway query trades error";
qryTrdErrCnt++;
qryTrdErrCnt--;
return 0;
}
Logger::getInstance().log->info("TradeClient::queryTrades ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
FbMsg::QryReqT queryT;
queryT.ChannelId = query.ChannelId;
queryT.AccountId = query.AccountId;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbQryReq = FbMsg::CreateQryReq(builder,&queryT);
builder.Finish(fbQryReq);


Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::QryTradesType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());

std::lock_guard<std::mutex> lock(m_qry_mtx);
ZmqHelper::sendPackMsg(m_pSendQueryReqSocket, &msg);

return 0;
}

int TradeClient::queryShareHolderInfo(QryReq& query)
{
if(!_checkGatewayStatus(query.ChannelId+query.AccountId)) // agent不在线
{
Logger::getInstance().log->error("TradeClient::queryShareHolderInfo client not online, gateway query shareholder error ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
static int qryShdErrCnt = 0;
if(qryShdErrCnt > STACK_DEEP)
{
return 0;
}
ErrorInfo error;
error.ChannelId = query.ChannelId;
error.AccountId = query.AccountId;
error.ErrorId = ERROR_ID_QRYSHARE;
error.ErrorMsg = "gateway query shareholder info error";
qryShdErrCnt++;
qryShdErrCnt--;
return 0;
}
Logger::getInstance().log->info("TradeClient::queryShareHolderInfo ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
FbMsg::QryReqT queryT;
queryT.ChannelId = query.ChannelId;
queryT.AccountId = query.AccountId;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbQryReq = FbMsg::CreateQryReq(builder,&queryT);
builder.Finish(fbQryReq);


Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::QryShareHolderType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());

std::lock_guard<std::mutex> lock(m_qry_mtx);
ZmqHelper::sendPackMsg(m_pSendQueryReqSocket, &msg);

return 0;
}

int TradeClient::heartBeat(HeartBeat &heart)
{
FbMsg::HeartBeatT heartT;
heartT.ChannelId = heart.ChannelId;
heartT.AccountId = heart.AccountId;
heartT.LastTime = heart.LastTime;
heartT.ErrorId = heart.ErrorId;
heartT.ErrorMsg = heart.ErrorMsg;


flatbuffers::FlatBufferBuilder builder(1024);
auto fbheart = FbMsg::CreateHeartBeat(builder,&heartT);
builder.Finish(fbheart);

Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::HeartBeatType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());

std::lock_guard<std::mutex> lock(m_trade_mtx);
ZmqHelper::sendPackMsg(m_pSendTradeReqSocket, &msg);
return 0;
}

void TradeClient::_sendHeartBeat()
{
while (true)
{
HeartBeat heart;
heart.ChannelId = m_channelid;
heart.AccountId = m_accountid;
heart.LastTime = TimeUtils::getCurrentTimestamp();
heart.ErrorId = 0;
heart.ErrorMsg = "OK";
heartBeat(heart);
Logger::getInstance().log->info("TradeClient::_sendHeartBeat ChannelId:{} AccountId:{}",heart.ChannelId,heart.AccountId);
std::this_thread::sleep_for(std::chrono::seconds(m_heartBeatInterval));
}
}

void TradeClient::_checkHeartBeatTimeout()
{
while(true)
{
long long last_time = m_last_heartbeat.load();
if(last_time > 0)
{
long long curtime = std::stoll(TimeUtils::getCurrentTimestamp());
long long dealt = curtime - last_time;
if(dealt > m_checkHeartBeatTimeout)
{
m_gateway_is_live.store(false);
Logger::getInstance().log->error("TradeClient::_checkHeartBeatTimeout channel:{} account:{} curtime:{} LastTime:{} dealt:{} heartBeatInterval:{} checkHeartBeatTimeout:{} connect agent timeout",m_channelid,m_accountid,curtime,last_time,dealt,m_heartBeatInterval,m_checkHeartBeatTimeout);
}
}
std::this_thread::sleep_for(std::chrono::seconds(1));
}
}

void TradeClient::_agentRegisterRsp(AgentInfo &agent)
{
FbMsg::AgentInfoT agentT;
agentT.ChannelId = agent.ChannelId;
agentT.AccountId = agent.AccountId;
agentT.LastTime = agent.LastTime;
agentT.Status = agent.Status;


flatbuffers::FlatBufferBuilder builder(1024);
auto fbagent = FbMsg::CreateAgentInfo(builder,&agentT);
builder.Finish(fbagent);

Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::AgentRegisterRspType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());

std::lock_guard<std::mutex> lock(m_trade_mtx);
ZmqHelper::sendPackMsg(m_pSendTradeReqSocket, &msg);
}

bool TradeClient::_checkGatewayStatus(std::string agentid)
{
return m_gateway_is_live.load();
}

void TradeClient::_createZmqCtx()
{
Logger::getInstance().log->debug("{}",GET_CLASS_FUNCTION_NAME);
m_pSendTradeReqContext = zmq_ctx_new();
m_pSendTradeReqSocket = zmq_socket(m_pSendTradeReqContext, ZMQ_DEALER);


m_pRecvTradeRtnContext = zmq_ctx_new();
m_pRecvTradeRtnSocket = zmq_socket(m_pRecvTradeRtnContext, ZMQ_DEALER);

m_pSendQueryReqContext = zmq_ctx_new();
m_pSendQueryReqSocket = zmq_socket(m_pSendQueryReqContext, ZMQ_DEALER);

m_pRecvQueryRtnContext = zmq_ctx_new();
m_pRecvQueryRtnSocket = zmq_socket(m_pRecvQueryRtnContext, ZMQ_DEALER);
}

void TradeClient::_setDealerSocket(ZmqSocketPtr socket, std::string identity, std::string addr)
{
Logger::getInstance().log->debug("{} identity:{} bind addr:{}",GET_CLASS_FUNCTION_NAME,identity,addr);
zmq_setsockopt(socket, ZMQ_IDENTITY, identity.c_str(), identity.length());
zmq_setsockopt(socket, ZMQ_TCP_KEEPALIVE, &m_tcp_keep_alive, sizeof(m_tcp_keep_alive));
zmq_setsockopt(socket, ZMQ_TCP_KEEPALIVE_IDLE, &m_tcp_keep_idle, sizeof(m_tcp_keep_idle));
zmq_setsockopt(socket, ZMQ_TCP_KEEPALIVE_INTVL, &m_tcp_keep_intvl, sizeof(m_tcp_keep_intvl));
zmq_setsockopt(socket, ZMQ_TCP_KEEPALIVE_CNT, &m_tcp_keep_cnt, sizeof(m_tcp_keep_cnt));
zmq_setsockopt(socket, ZMQ_HEARTBEAT_IVL, &m_heartbeat_ivl, sizeof(m_heartbeat_ivl));
zmq_setsockopt(socket, ZMQ_HEARTBEAT_TIMEOUT, &m_heartbeat_timeout, sizeof(m_heartbeat_timeout));
zmq_setsockopt(socket, ZMQ_HEARTBEAT_TTL, &m_heartbeat_ttl, sizeof(m_heartbeat_ttl));
zmq_connect(socket, addr.c_str());
}

bool TradeClient::_checkClientStatus(std::string clientid)
{
return false;
}
使用C# clrzmq2+ flatbuffer 方式， 先完整转账发送心跳包的代码，可直接运行