modularization/src/driver/driver_rpc_server/main.cpp

#include <QCoreApplication>

#include <yaml-cpp/yaml.h>
#include <brpc/server.h>
#include "echo.pb.h"

#include "modulecomm.h"

#include <thread>

#include "ivfault.h"
#include "ivlog.h"
#include "ivversion.h"
#include "ivbacktrace.h"

iv::Ivfault *gfault = nullptr;
iv::Ivlog *givlog = nullptr;

namespace iv {
struct msgunit
{
    char mstrmsgname[256];
    int mnBufferSize = 10000;
    int mnBufferCount = 1;
    void * mpa;
};
}

std::vector<iv::msgunit> mvectormsgunit;

void * gpa;

unsigned int gport = 60031;

namespace iv {
class EchoServiceImpl : public EchoService {
public:
    EchoServiceImpl() {};
    virtual ~EchoServiceImpl() {};
    virtual void Echo(google::protobuf::RpcController* cntl_base,
                      const EchoRequest* request,
                      EchoResponse* response,
                      google::protobuf::Closure* done) {
        // This object helps you to call done->Run() in RAII style. If you need
        // to process the request asynchronously, pass done_guard.release().
        brpc::ClosureGuard done_guard(done);

        brpc::Controller* cntl =
            static_cast<brpc::Controller*>(cntl_base);

        // The purpose of following logs is to help you to understand
        // how clients interact with servers more intuitively. You should
        // remove these logs in performance-sensitive servers.
//        LOG(INFO) << "Received request[log_id=" << cntl->log_id()
//                  << "] from " << cntl->remote_side()
//                  << " to " << cntl->local_side()
//                  << ": " << request->msgname()
//                  << " (attached=" << cntl->request_attachment() << ")";


//        givlog->debug("ECHOSTATE","HELLO");
        givlog->debug("ECHOSTATE","Received request[log_id=%d] from%d to %d: %s",cntl->log_id(),cntl->remote_side(),cntl->local_side(),request->msgname().data());


        // Fill response.
        response->set_msgname(request->msgname());
        response->set_index(request->index());



        int i;
        for(i=0;i<mvectormsgunit.size();i++)
        {
            if(strncmp(request->msgname().data(),mvectormsgunit[i].mstrmsgname,255) == 0)
            {
                iv::modulecomm::ModuleSendMsg(mvectormsgunit[i].mpa,request->xdata().data(),request->xdata().length());
                break;
            }
        }

        // You can compress the response by setting Controller, but be aware
        // that compression may be costly, evaluate before turning on.
        // cntl->set_response_compress_type(brpc::COMPRESS_TYPE_GZIP);

        if (true) {
            // Set attachment which is wired to network directly instead of
            // being serialized into protobuf messages.
            cntl->response_attachment().append(cntl->request_attachment());
        }
    }
};

}



void dec_yaml(const char * stryamlpath)
{

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

    int i;
    int nmodulesize;
    std::vector<std::string> vecmodulename;


    if(config["port"])
    {
        int nport = config["port"].as<int>();
        gport = nport;
    }
    if(config["message"])
    {
        nmodulesize = config["message"].size();
        for(i=0;i<nmodulesize;i++)
        {
            std::string strname = config["message"][i].as<std::string>();
            vecmodulename.push_back(strname);
        }
    }
    else
    {
        return;
    }

    if(nmodulesize <1)return;

    std::string strmsgname;

    for(i=0;i<nmodulesize;i++)
    {
        if(config[vecmodulename[i].data()])
        {
            if(config[vecmodulename[i].data()]["msgname"])
            {
                strmsgname = config[vecmodulename[i].data()]["msgname"].as<std::string>();

                iv::msgunit xmu;
                strncpy(xmu.mstrmsgname,strmsgname.data(),255);

                if(config[vecmodulename[i].data()]["buffersize"])
                {
                    xmu.mnBufferSize = config[vecmodulename[i].data()]["buffersize"].as<int>();
                }
                if(config[vecmodulename[i].data()]["buffercount"])
                {
                    xmu.mnBufferCount = config[vecmodulename[i].data()]["buffercount"].as<int>();
                }
                mvectormsgunit.push_back(xmu);
            }
        }
    }




    return;

}


void rpcthread()
{
    // Generally you only need one Server.
    brpc::Server server;

    // Instance of your service.
    iv::EchoServiceImpl echo_service_impl;

    // Add the service into server. Notice the second parameter, because the
    // service is put on stack, we don't want server to delete it, otherwise
    // use brpc::SERVER_OWNS_SERVICE.
    if (server.AddService(&echo_service_impl,
                          brpc::SERVER_DOESNT_OWN_SERVICE) != 0) {
        std::cout << "Fail to add service" << std::endl;
        return ;
    }

    // Start the server.
    brpc::ServerOptions options;
    options.idle_timeout_sec = -1;
    if (server.Start(gport, &options) != 0) {
        std::cout << "Fail to start EchoServer"<<std::endl;
        return ;
    }

    // Wait until Ctrl-C is pressed, then Stop() and Join() the server.
    server.RunUntilAskedToQuit();
}

int main(int argc, char *argv[])
{
    RegisterIVBackTrace();
    showversion("driver_rpc_server");
    QCoreApplication a(argc, argv);


    gfault = new iv::Ivfault("driver_rpc_server");
    givlog = new iv::Ivlog("driver_rpc_server");

    char stryamlpath[256];
    if(argc<2)
    {
        snprintf(stryamlpath,255,"driver_rpc_server.yaml");
//        strncpy(stryamlpath,abs_ymlpath,255);
    }
    else
    {
        strncpy(stryamlpath,argv[1],255);
    }
    dec_yaml(stryamlpath);

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

    std::thread * pthread = new std::thread(rpcthread);



    return a.exec();
}