modularization/src/driver/driver_cloud_grpc_pc_stream/grpcpc.cpp

#include "grpcpc.h"


static grpcpc * ggrpcpc;

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


grpcpc::grpcpc(std::string stryamlpath)
{
    ggrpcpc = this;
    dec_yaml(stryamlpath.data());
    int i;
    for(i=0;i<mvectormsgunit.size();i++)
    {
        mvectormsgunit[i].mpa = iv::modulecomm::RegisterSend(mvectormsgunit[i].mstrmsgname,mvectormsgunit[i].mnBufferSize,mvectormsgunit[i].mnBufferCount);
    }

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


void grpcpc::setframerate(float framerate)
{
    mframerate = framerate;
}

void grpcpc::threadsend(std::shared_ptr<::grpc::ClientReaderWriter<iv::queryReqStream, iv::queryReplyStream> > writer, bool *pbrun)
{
    int nctrlsize = mvectorctrlmsgunit.size();
    int i;

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

    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)<ninterval)
        {
            continue;
        }

        iv::queryReqStream request;

        bool bImportant = false;
        int nkeeptime = 0;

            iv::cloud::cloudmsg xmsg;
            xmsg.set_xtime(QDateTime::currentMSecsSinceEpoch());
            nlastsend = xTime.elapsed();
            gMutexMsg.lock();

  //          std::vector<iv::msgunit> xv = mvectorctrlmsgunit;
            for(i=0;i<nctrlsize;i++)
            {
                if(mvectorctrlmsgunit[i].mbRefresh)
                {
                    mvectorctrlmsgunit[i].mbRefresh = false;

                    if(mvectorctrlmsgunit[i].mbImportant)
                    {
                        bImportant = true;
                    }
                    if(mvectorctrlmsgunit[i].mnkeeptime > nkeeptime)
                    {
                        nkeeptime = mvectorctrlmsgunit[i].mnkeeptime;
                    }
                    iv::cloud::cloudunit xcloudunit;
                    xcloudunit.set_msgname(mvectorctrlmsgunit[i].mstrmsgname);
                    xcloudunit.set_data(mvectorctrlmsgunit[i].mpstrmsgdata.get(),mvectorctrlmsgunit[i].mndatasize);
                    iv::cloud::cloudunit * pcu = xmsg.add_xclouddata();
                    pcu->CopyFrom(xcloudunit);
                }

            }
            gMutexMsg.unlock();

            request.set_strquerymd5(gstrqueryMD5);
            request.set_strvin(gstrVIN);

//            request.set_ntime(QDateTime::currentMSecsSinceEpoch());
            request.set_ntype(0);

            if(xmsg.xclouddata_size()>0)
            {
                int nbytesize = xmsg.ByteSize();
                std::vector<char> pvectordata;
                pvectordata.resize(nbytesize);
                if(xmsg.SerializeToArray(pvectordata.data(),nbytesize))
                {

//                    request.set_id(nctrlid);nctrlid++;
                    request.set_strctrlmd5(gstrctrlMD5);
                    request.set_nsuggestframerate(mframerate);
                    request.set_xdata(pvectordata.data(),pvectordata.size());
                    request.set_bimportant(bImportant);
                    request.set_kepptime(nkeeptime);
                    request.set_ntype(1);
                }
            }
            nlastsend = xTime.elapsed();

            bool bsend = writer->Write(request);
            std::cout<<"send msg. rtn is "<<bsend<<std::endl;

    }
}

void grpcpc::run()
{
    int nsize = mvectormsgunit.size();
    int nctrlsize = mvectorctrlmsgunit.size();
    int i;

    qint64 nlasttime = 0;

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

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

    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<::grpc::ClientReaderWriter<iv::queryReqStream, iv::queryReplyStream> > writerRead(stub_->queryctrl(&context));

        bool bRun = true;
        std::thread * pthread = new std::thread(&grpcpc::threadsend,this,writerRead,&bRun);
        (void )pthread;
        iv::queryReplyStream reply;
        while (writerRead->Read(&reply)) {
            nfail = 0;
    //        std::cout << "接收到回复：" << reply.remsg()<<"--\n" << std::endl;
            if(reply.nres() == 1)
            {
                std::cout<<"data size is "<<reply.xdata().size()<<"time is"<<QDateTime::currentMSecsSinceEpoch()<<std::endl;
                iv::cloud::cloudmsg xmsg;
                if(xmsg.ParseFromArray(reply.xdata().data(),reply.xdata().size()))
                {
                    sharequerymsg(&xmsg);
                }
            }
            std::cout<<"read data from server. nres :"<< reply.nres()<<std::endl;
        }
        bRun = false;
        pthread->join();
        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 grpcpc::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>();
    }

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


    std::string strmsgname;

    if(config["querymessage"])
    {

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

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


            }
        }
    }
    else
    {

    }

    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>();

                if(config[strnodename][strtitle]["bimportant"])
                {
                   std::string strimportant =    config[strnodename][strtitle]["bimportant"].as<std::string>();
                   if(strimportant == "true")
                   {
                       xmu.mbImportant = true;
                   }
                }
                if(config[strnodename][strtitle]["keeptime"])
                {
                   std::string strkeep =    config[strnodename][strtitle]["keeptime"].as<std::string>();
                   xmu.mnkeeptime = atoi(strkeep.data());
                }
                mvectorctrlmsgunit.push_back(xmu);
            }
        }
    }
    else
    {

    }

    return;
}

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

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

std::string grpcpc::GetVIN()
{
    return gstrVIN;
}

float grpcpc::getframerate()
{
    return mframerate;
}