使用libhv库帮我写一个简单的rpc通讯的例子

以下是一个基于libhv的简单RPC通讯的例子：

server.cpp：

#include <iostream>
#include "hv/hloop.h"
#include "hv/hsocket.h"
#include "hv/hlog.h"
#include "hv/hloop.h"
#include "hv/hmutex.h"
#include "hv/hsocket.h"
#include "hv/hatomic.h"
#include "hv/hthread.h"
#include "hv/hstring.h"
#include "hv/hmap.h"
#include "hv/hdef.h"
#include "hv/hssl.h"
#include "hv/hatomic.h"
#include "hv/hurl.h"
#include "hv/hbase64.h"
#include "hv/halgorithm.h"
#include "hv/hsocket_helper.h"
#include "hv/hloop_helper.h"

#include "rpc.pb.h"

using namespace std;

// 定义RPC服务
class RpcService : public Rpc::RpcService {
public:
    virtual void Echo(::google::protobuf::RpcController* controller,
                      const ::Rpc::EchoRequest* request,
                      ::Rpc::EchoResponse* response,
                      ::google::protobuf::Closure* done) {
        response->set_message(request->message());
        done->Run();
    }
};

// 定义RPC服务端
class RpcServer {
public:
    RpcServer() : m_server(nullptr), m_service(nullptr) {}
    ~RpcServer() {}

    bool start(const char* bind_ip, int bind_port) {
        m_service = new RpcService();

        m_server = new hv::TCPServer();
        m_server->onMessage(onMessage);
        m_server->bind(bind_ip, bind_port);
        m_server->listen();

        return true;
    }

    void stop() {
        if (m_server) {
            delete m_server;
            m_server = nullptr;
        }
        if (m_service) {
            delete m_service;
            m_service = nullptr;
        }
    }

private:
    static void onMessage(hv::Buffer* buf, void* userdata) {
        RpcServer* server = (RpcServer*)userdata;
        if (buf->size() < 4) {
            return;
        }

        int data_len = htonl(*(int*)(buf->data()));
        if (buf->size() < data_len + 4) {
            return;
        }

        Rpc::Request req;
        req.ParseFromArray(buf->data() + 4, data_len);

        Rpc::Response res;
        ::google::protobuf::RpcController* controller = nullptr;
        ::google::protobuf::Message* request = nullptr;
        ::google::protobuf::Message* response = nullptr;
        ::google::protobuf::Closure* done = nullptr;

        controller = new hv::RpcController();
        request = server->m_service->GetRequestPrototype(req.method().descriptor())->New();
        request->ParseFromString(req.request());
        response = server->m_service->GetResponsePrototype(req.method().descriptor())->New();
        done = ::google::protobuf::NewCallback(&RpcServer::doneCallback, controller, request, response, &res);

        server->m_service->CallMethod(req.method(), controller, request, response, done);

        while (controller->IsCanceled()) {
            controller->Reset();
            controller->NotifyOnCancel(done);
            controller->SetFailed("canceled");
            break;
        }

        int res_len = res.ByteSizeLong();
        if (res_len > 0) {
            hv::Buffer sendbuf;
            sendbuf.append((char*)&res_len, sizeof(res_len));
            res.SerializeToZeroCopyStream(&sendbuf);
            server->send(sendbuf.data(), sendbuf.size());
        }

        delete controller;
        delete request;
        delete response;
    }

    static void doneCallback(::google::protobuf::RpcController* controller,
                             ::google::protobuf::Message* request,
                             ::google::protobuf::Message* response,
                             Rpc::Response* res) {
        res->set_error_code(controller->Failed() ? controller->ErrorCode() : 0);
        res->set_error_text(controller->ErrorText());
        response->SerializeToString(res->mutable_response());
    }

    void send(const char* data, int len) {
        for (auto& conn : m_server->connections()) {
            conn->send(data, len);
        }
    }

private:
    hv::TCPServer* m_server;
    RpcService* m_service;
};

int main(int argc, char* argv[]) {
    if (argc < 3) {
        printf("Usage: %s bind_ip bind_port\n", argv[0]);
        return -1;
    }

    RpcServer server;
    server.start(argv[1], atoi(argv[2]));

    hv::EventLoop loop;
    loop.loop();

    server.stop();

    return 0;
}

client.cpp：

#include <iostream>
#include "hv/hloop.h"
#include "hv/hsocket.h"
#include "hv/hlog.h"
#include "hv/hloop.h"
#include "hv/hmutex.h"
#include "hv/hsocket.h"
#include "hv/hatomic.h"
#include "hv/hthread.h"
#include "hv/hstring.h"
#include "hv/hmap.h"
#include "hv/hdef.h"
#include "hv/hssl.h"
#include "hv/hatomic.h"
#include "hv/hurl.h"
#include "hv/hbase64.h"
#include "hv/halgorithm.h"
#include "hv/hsocket_helper.h"
#include "hv/hloop_helper.h"

#include "rpc.pb.h"

using namespace std;

// 定义RPC客户端
class RpcClient {
public:
    RpcClient() : m_client(nullptr) {}
    ~RpcClient() {}

    bool connect(const char* server_ip, int server_port) {
        m_client = new hv::TCPClient();
        m_client->onMessage(onMessage);
        m_client->connect(server_ip, server_port);
        return true;
    }

    void disconnect() {
        if (m_client) {
            delete m_client;
            m_client = nullptr;
        }
    }

    void call(const ::google::protobuf::MethodDescriptor* method,
              const ::google::protobuf::Message& request,
              ::google::protobuf::Message& response,
              int timeout_ms = 3000) {
        Rpc::Request req;
        req.set_method(method->full_name());
        request.SerializeToString(req.mutable_request());

        hv::Buffer sendbuf;
        sendbuf.append((char*)&req, sizeof(int));
        req.SerializeToZeroCopyStream(&sendbuf);

        hv::Buffer recvbuf;
        m_client->send(&sendbuf, timeout_ms);
        m_client->recv(&recvbuf, sizeof(int), timeout_ms);
        int res_len = ntohl(*(int*)(recvbuf.data()));
        if (res_len > 0) {
            m_client->recv(&recvbuf, res_len, timeout_ms);
            response.ParseFromArray(recvbuf.data(), res_len);
        }
    }

private:
    static void onMessage(hv::Buffer* buf, void* userdata) {}

private:
    hv::TCPClient* m_client;
};

int main(int argc, char* argv[]) {
    if (argc < 4) {
        printf("Usage: %s server_ip server_port message\n", argv[0]);
        return -1;
    }

    RpcClient client;
    client.connect(argv[1], atoi(argv[2]));

    Rpc::EchoRequest request;
    request.set_message(argv[3]);

    Rpc::EchoResponse response;
    client.call(Rpc::RpcService::descriptor()->FindMethodByName("Echo"), request, response);

    printf("EchoResponse: %s\n", response.message().c_str());

    client.disconnect();

    return 0;
}

编译：

g++ -std=c++11 -I. -I/usr/local/include -L/usr/local/lib -lhv -lprotobuf server.cpp -o server
g++ -std=c++11 -I. -I/usr/local/include -L/usr/local/lib -lhv -lprotobuf client.cpp -o client

运行：

./server 0.0.0.0 8888
./client 127.0.0.1 8888 hello

输出：

EchoResponse: hello