modularization/src/driver/driver_cloud_grpc_client_stream/grpcclient.cpp

#include "grpcclient.h"

grpcclient * ggrpcclient;

void ListenData(const char * strdata,const unsigned int nSize,const unsigned int index,const QDateTime * dt,const char * strmemname)
{
    ggrpcclient->UpdateData(strdata,nSize,strmemname);

}


grpcclient::grpcclient(std::string stryamlpath)
{

    ggrpcclient = this;
    dec_yaml(stryamlpath.data());

    int i;
    for(i=0;i<mvectormsgunit.size();i++)
    {
        mvectormsgunit[i].mpa = iv::modulecomm::RegisterRecv(mvectormsgunit[i].mstrmsgname,ListenData);
    }

    for(i=0;i<mvectorctrlmsgunit.size();i++)
    {
        mvectorctrlmsgunit[i].mpa = iv::modulecomm::RegisterSend(mvectorctrlmsgunit[i].mstrmsgname,mvectorctrlmsgunit[i].mnBufferSize,
                                                                 mvectorctrlmsgunit[i].mnBufferCount);
    }
}

qint64 grpcclient::calclatency(qint64 nnewlatency)
{
    mvectorlatency.push_back(nnewlatency);
    while(mvectorlatency.size()>5)mvectorlatency.erase(mvectorlatency.begin());
    qint64 nlatencytotal =0;
    int nsize = mvectorlatency.size();
    int i;
    for(i=0;i<nsize;i++)
    {
        nlatencytotal = nlatencytotal + mvectorlatency.at(i);
    }
    if(nsize > 0)
    {
        nlatencytotal = nlatencytotal/nsize;
    }
    mnlatency = nlatencytotal;
    return nlatencytotal;
}

void grpcclient::threadsend(std::shared_ptr<::grpc::ClientReaderWriter<iv::UploadRequestStream, iv::UploadReplyStream> > writer,std::shared_ptr<bool> pbrun,std::shared_ptr<qint64> nlastreftime)
{
    std::cout<<"threadsend start "<<std::endl;
    int nsize = mvectormsgunit.size();
    int i;

    int ninterval = atoi(gstruploadinterval.data());
    if(ninterval<=0)ninterval = 100;

    mninterval = ninterval;
    float ffraterate = 1.0f/((float)mninterval);
    int nrawinterval = ninterval;
    int nok = 0;

    QTime xTime;
    xTime.start();
    int nlastsend = xTime.elapsed();

    int nid= 0;

    while(*pbrun)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
        if((xTime.elapsed()-nlastsend)<mninterval)
        {
            continue;
        }

        bool bImportant = false;
        int nkeeptime = 0;
        iv::cloud::cloudmsg xmsg;
        xmsg.set_xtime(QDateTime::currentMSecsSinceEpoch());
        gMutexMsg.lock();
        for(i=0;i<nsize;i++)
        {
            if(mvectormsgunit[i].mbRefresh)
            {
                mvectormsgunit[i].mbRefresh = false;
                if(mvectormsgunit[i].mbImportant)
                {
                    bImportant = true;
                }
                if(mvectormsgunit[i].mnkeeptime > nkeeptime)
                {
                    nkeeptime = mvectormsgunit[i].mnkeeptime;
                }
                iv::cloud::cloudunit xcloudunit;
                xcloudunit.set_msgname(mvectormsgunit[i].mstrmsgname);
                xcloudunit.set_data(mvectormsgunit[i].mpstrmsgdata.get(),mvectormsgunit[i].mndatasize);
                iv::cloud::cloudunit * pcu = xmsg.add_xclouddata();
                pcu->CopyFrom(xcloudunit);
            }

        }
        gMutexMsg.unlock();

        int nbytesize = xmsg.ByteSize();
        char * strbuf = new char[nbytesize];
        std::shared_ptr<char> pstrbuf;
        pstrbuf.reset(strbuf);
        if(xmsg.SerializeToArray(strbuf,nbytesize))
        {
            iv::UploadRequestStream request;
            qint64 time1 = QDateTime::currentMSecsSinceEpoch();

            request.set_id(nid);
            request.set_ntime(time1);
            request.set_strquerymd5(gstrqueryMD5);
            request.set_strctrlmd5(gstrctrlMD5);
            request.set_strvin(gstrVIN);
            request.set_xdata(strbuf,nbytesize);
            request.set_kepptime(nkeeptime);
            request.set_bimportant(bImportant);
            request.set_nlatency(mnlatency);
            ffraterate = 1000.0f/((float)mninterval);
            request.set_fframerate(ffraterate);
            request.set_nsendtime(time1);
            nid++;

            nlastsend = xTime.elapsed();

            QTime xt;
            xt.start();
            ::grpc::WriteOptions wo;
//            wo.set_write_through();
//            wo.clear_buffer_hint();
    //        writer->Write(request,(void * )2);

    //        bool bsend = true;
            bool bsend = writer->Write(request,wo);

            *nlastreftime = QDateTime::currentMSecsSinceEpoch();

//            if(xt.elapsed()>10)
//            {
//                nok = 0;
//                if(ninterval < 1000)ninterval = ninterval * 11/10;
//                mninterval = ninterval;
//                ffraterate = 1.0f/((float)mninterval);
//                qDebug("send ela is %d ninterval is %d",xt.elapsed(),ninterval);
//            }
//            else
//            {
//                nok++;
//                if((ninterval > nrawinterval)&&(nok>10))
//                {
//                    nok = 0;

//                    ninterval = ninterval*10/11;
//                    mninterval = ninterval;
//                    ffraterate = 1.0f/((float)mninterval);
//                    std::cout<<"ninterval is "<<ninterval<<std::endl;
//                }
//            }
            if(bsend == false)std::cout<<"send msg. rtn is "<<bsend<<std::endl;


        }


    }
    std::cout<<"thread send end."<<std::endl;
}



void grpcclient::threadrecv(std::shared_ptr<::grpc::ClientReaderWriter<iv::UploadRequestStream, iv::UploadReplyStream> > writer, std::shared_ptr<bool> pbrun,std::shared_ptr<qint64> nlastreftime)
{
    std::cout<<"threadrecv start"<<std::endl;
    iv::UploadReplyStream reply;
    while (writer->Read(&reply)) {

        *nlastreftime = QDateTime::currentMSecsSinceEpoch();
        qint64 nlaten = QDateTime::currentMSecsSinceEpoch() - reply.nreqsendtime();
        if(reply.nreqsendtime() == 0)nlaten = 0;
        else
        {
            nlaten = nlaten - reply.npausetime();
        }
        calclatency(nlaten);

        if(reply.framerate() >0.001)
        {
            mninterval = 1000.0/reply.framerate();
        }

        qDebug("latency is %ld",nlaten);
//        nfail = 0;
//        std::cout << "接收到回复：" << reply.remsg()<<"--\n" << std::endl;
        if(reply.nres() == 1)
        {
            iv::cloud::cloudmsg xmsg;
            if(xmsg.ParseFromArray(reply.xdata().data(),reply.xdata().size()))
            {
                sharectrlmsg(&xmsg);
            }
        }

//            std::cout<<"read data from server."<<std::endl;
    }
    std::cout<<"threadrecv end."<<std::endl;
    *pbrun = false;
}


void grpcclient::threadRPC(std::shared_ptr<qint64> pnrpctime,std::shared_ptr<bool> pbRun)
{
    std::cout<<"threadrpc start."<<std::endl;
    std::string target_str = gstrserverip+":";
    target_str = target_str + gstrserverport ;//std::to_string()
    auto cargs = grpc::ChannelArguments();
    cargs.SetMaxReceiveMessageSize(1024 * 1024 * 1024); // 1 GB
    cargs.SetMaxSendMessageSize(1024 * 1024 * 1024);

    std::shared_ptr<Channel> channel = grpc::CreateCustomChannel(
             target_str, grpc::InsecureChannelCredentials(),cargs);

    std::unique_ptr<iv::UploadStream::Stub> stub_ = iv::UploadStream::NewStub(channel);


    int nfail = 0;

    while(!QThread::isInterruptionRequested())
    {
        ClientContext context ;
//        std::shared_ptr<ClientContext> pcontext(new ClientContext);
        std::shared_ptr<::grpc::ClientReaderWriter<iv::UploadRequestStream, iv::UploadReplyStream> > writerRead(stub_->upload(&context));


//        std::shared_ptr<bool> pbRun(new bool);
        *pbRun = true;
        std::shared_ptr<qint64> pntime = pnrpctime ;
        *pntime = QDateTime::currentMSecsSinceEpoch();
        std::thread * pthread = new std::thread(&grpcclient::threadsend,this,writerRead,pbRun,pntime);
        (void )pthread;
        std::thread * precvthread = new std::thread(&grpcclient::threadrecv,this,writerRead,pbRun,pntime);
        (void)precvthread;

        pthread->join();
        precvthread->join();

 //       std::cout<<"threadRPC end"<<std::endl;
 //       *pbRun = false;

//        while(((QDateTime::currentMSecsSinceEpoch() - *pntime)<3000)&&(*pbRun))
//        {
//            std::this_thread::sleep_for(std::chrono::milliseconds(1000));
//        }

//        *pbRun = false;



        std::this_thread::sleep_for(std::chrono::milliseconds(1000));

        channel = grpc::CreateCustomChannel(
                 target_str, grpc::InsecureChannelCredentials(),cargs);
        stub_ = iv::UploadStream::NewStub(channel);
        nfail++;
//        if(nfail > 100)std::this_thread::sleep_for(std::chrono::milliseconds(3000));
//        else std::this_thread::sleep_for(std::chrono::milliseconds(100));
        std::cout<<"reconnnect to server. nfail is "<<nfail<<std::endl;
    }
}

void grpcclient::run()
{
            std::shared_ptr<bool> pbRun(new bool);
            *pbRun = true;
                    std::shared_ptr<qint64> pntime(new qint64);
                    *pntime = QDateTime::currentMSecsSinceEpoch();

    std::thread * pthread = new std::thread(&grpcclient::threadRPC,this,pntime,pbRun);
    return;

//    while(!QThread::isInterruptionRequested())
//    {
//        std::shared_ptr<bool> pbRun(new bool);
//        *pbRun = true;
//        std::shared_ptr<qint64> pntime(new qint64);
//        *pntime = QDateTime::currentMSecsSinceEpoch();
//        std::thread * pthread = new std::thread(&grpcclient::threadRPC,this,pntime,pbRun);
//        (void )pthread;
//        while(((QDateTime::currentMSecsSinceEpoch() - *pntime)<10000)&&(*pbRun))
//        {
//            std::this_thread::sleep_for(std::chrono::milliseconds(1000));
//        }
//        std::cout<<" reconect."<<std::endl;
//        *pbRun = false;

//    }

}


void grpcclient::dec_yaml(const char * stryamlpath)
{

    YAML::Node config;
    try
    {
        config = YAML::LoadFile(stryamlpath);
    }
    catch(YAML::BadFile e)
    {
        qDebug("load error.");
        return;
    }

    std::vector<std::string> vecmodulename;


    if(config["server"])
    {
        gstrserverip = config["server"].as<std::string>();
    }
    if(config["port"])
    {
        gstrserverport = config["port"].as<std::string>();
    }
    if(config["uploadinterval"])
    {
        gstruploadinterval = config["uploadinterval"].as<std::string>();
    }

    if(config["VIN"])
    {
        gstrVIN = config["VIN"].as<std::string>();
    }

    if(config["queryMD5"])
    {
        gstrqueryMD5 = config["queryMD5"].as<std::string>();
    }
    else
    {
        return;
    }

    if(config["ctrlMD5"])
    {
        gstrctrlMD5 = config["ctrlMD5"].as<std::string>();
    }


    std::string strmsgname;

    if(config["uploadmessage"])
    {

        for(YAML::const_iterator it= config["uploadmessage"].begin(); it != config["uploadmessage"].end();++it)
        {
            std::string strtitle = it->first.as<std::string>();
            std::cout<<strtitle<<std::endl;

            if(config["uploadmessage"][strtitle]["msgname"]&&config["uploadmessage"][strtitle]["buffersize"]&&config["uploadmessage"][strtitle]["buffercount"])
            {
                iv::msgunit xmu;
                strmsgname = config["uploadmessage"][strtitle]["msgname"].as<std::string>();
                strncpy(xmu.mstrmsgname,strmsgname.data(),255);
                xmu.mnBufferSize = config["uploadmessage"][strtitle]["buffersize"].as<int>();
                xmu.mnBufferCount = config["uploadmessage"][strtitle]["buffercount"].as<int>();
                if(config["uploadmessage"][strtitle]["bimportant"])
                {
                   std::string strimportant =    config["uploadmessage"][strtitle]["bimportant"].as<std::string>();
                   if(strimportant == "true")
                   {
                       xmu.mbImportant = true;
                   }
                }
                if(config["uploadmessage"][strtitle]["keeptime"])
                {
                   std::string strkeep =    config["uploadmessage"][strtitle]["keeptime"].as<std::string>();
                   xmu.mnkeeptime = atoi(strkeep.data());
                }
                mvectormsgunit.push_back(xmu);
            }
        }
    }
    else
    {


    }

    if(!config["ctrlMD5"])
    {
       return;
    }

    if(config["ctrlmessage"])
    {
        std::string strnodename = "ctrlmessage";
        for(YAML::const_iterator it= config[strnodename].begin(); it != config[strnodename].end();++it)
        {
            std::string strtitle = it->first.as<std::string>();
            std::cout<<strtitle<<std::endl;

            if(config[strnodename][strtitle]["msgname"]&&config[strnodename][strtitle]["buffersize"]&&config[strnodename][strtitle]["buffercount"])
            {
                iv::msgunit xmu;
                strmsgname = config[strnodename][strtitle]["msgname"].as<std::string>();
                strncpy(xmu.mstrmsgname,strmsgname.data(),255);
                xmu.mnBufferSize = config[strnodename][strtitle]["buffersize"].as<int>();
                xmu.mnBufferCount = config[strnodename][strtitle]["buffercount"].as<int>();
                mvectorctrlmsgunit.push_back(xmu);
            }
        }
    }
    else
    {

    }

    return;

}

void grpcclient::sharectrlmsg(iv::cloud::cloudmsg * pxmsg)
{
    int i;
    int nsize = pxmsg->xclouddata_size();
    for(i=0;i<nsize;i++)
    {
        int j;
        int nquerysize = mvectorctrlmsgunit.size();
        for(j=0;j<nquerysize;j++)
        {
            if(strncmp(pxmsg->xclouddata(i).msgname().data(), mvectorctrlmsgunit[j].mstrmsgname,255) == 0)
            {
 //               qDebug("size is %d ",pxmsg->xclouddata(i).data().size());
                iv::modulecomm::ModuleSendMsg(mvectorctrlmsgunit[j].mpa,pxmsg->xclouddata(i).data().data(),pxmsg->xclouddata(i).data().size());
                break;
            }
        }
    }
}

void grpcclient::UpdateData(const char *strdata, const unsigned int nSize, const char *strmemname)
{
    int nsize = mvectormsgunit.size();
    int i;
    for(i=0;i<nsize;i++)
    {
        if(strncmp(strmemname,mvectormsgunit[i].mstrmsgname,255) == 0)
        {
            gMutexMsg.lock();
            char * strtem = new char[nSize];
            memcpy(strtem,strdata,nSize);
            mvectormsgunit[i].mpstrmsgdata.reset(strtem);
            mvectormsgunit[i].mndatasize = nSize;
            mvectormsgunit[i].mbRefresh = true;
            gMutexMsg.unlock();
            break;
        }
    }
}