kvraft_client.cpp

/* kvraft_client.cpp

 Generated by phxrpc_pb2client from kvraft.proto

*/

#include "kvraft_client.h"

#include <cstdlib>
#include <memory>
#include <mutex>

#include "phxrpc/http.h"
#include "phxrpc/rpc.h"

#include "phxrpc_kvraft_stub.h"


using namespace std;


static phxrpc::ClientConfig global_kvraftclient_config_;
static phxrpc::ClientMonitorPtr global_kvraftclient_monitor_;


bool KVRaftClient::Init(const char *config_file) {
    return global_kvraftclient_config_.Read(config_file);
}

const char *KVRaftClient::GetPackageName() {
    const char *ret = global_kvraftclient_config_.GetPackageName();
    if (strlen(ret) == 0) {
        ret = "kvraft";
    }
    return ret;
}

KVRaftClient::KVRaftClient() {
    static mutex monitor_mutex;
    if (!global_kvraftclient_monitor_.get()) {
        monitor_mutex.lock();
        if (!global_kvraftclient_monitor_.get()) {
            global_kvraftclient_monitor_ = phxrpc::MonitorFactory::GetFactory()->
                    CreateClientMonitor(GetPackageName());
        }
        global_kvraftclient_config_.SetClientMonitor(global_kvraftclient_monitor_);
        monitor_mutex.unlock();
    }
}

KVRaftClient::~KVRaftClient() {
}

int KVRaftClient::PHXEcho(const google::protobuf::StringValue &req, google::protobuf::StringValue *resp)
{
    const phxrpc::Endpoint_t *ep{global_kvraftclient_config_.GetRandom()};

    if (ep) {
        phxrpc::BlockTcpStream socket;
        bool open_ret{phxrpc::PhxrpcTcpUtils::Open(&socket, ep->ip, ep->port,
                global_kvraftclient_config_.GetConnectTimeoutMS(), nullptr, 0,
                *(global_kvraftclient_monitor_.get()))};
        if (open_ret) {
            socket.SetTimeout(global_kvraftclient_config_.GetSocketTimeoutMS());
            phxrpc::HttpMessageHandlerFactory http_msg_factory;
            KVRaftStub stub(socket, *(global_kvraftclient_monitor_.get()), http_msg_factory);
            return stub.PHXEcho(req, resp);
        }

    }

    return -1;
}

int KVRaftClient::PHXBatchEcho(const google::protobuf::StringValue &req, google::protobuf::StringValue *resp)
{
    int ret{-1};
    size_t echo_server_count{2};
    uthread_begin;
    for (size_t i{0}; echo_server_count > i; ++i) {
        uthread_t [=, &uthread_s, &ret](void *) {
            const phxrpc::Endpoint_t *ep = global_kvraftclient_config_.GetByIndex(i);
            if (ep != nullptr) {
                phxrpc::UThreadTcpStream socket;
                if (phxrpc::PhxrpcTcpUtils::Open(&uthread_s, &socket, ep->ip, ep->port,
                    global_kvraftclient_config_.GetConnectTimeoutMS(), *(global_kvraftclient_monitor_.get()))) {
                    socket.SetTimeout(global_kvraftclient_config_.GetSocketTimeoutMS());
                    phxrpc::HttpMessageHandlerFactory http_msg_factory;
                    KVRaftStub stub(socket, *(global_kvraftclient_monitor_.get()), http_msg_factory);
                    int this_ret{stub.PHXEcho(req, resp)};
                    if (this_ret == 0) {
                        ret = this_ret;
                        uthread_s.Close();
                    }
                }
            }
        };
    }
    uthread_end;
    return ret;
}

int KVRaftClient::RequestVote(const kvraft::RequestVoteArgs &req, kvraft::RequestVoteReply *resp)
{
    const phxrpc::Endpoint_t *ep{global_kvraftclient_config_.GetRandom()};

    if (ep) {
        phxrpc::BlockTcpStream socket;
        bool open_ret{phxrpc::PhxrpcTcpUtils::Open(&socket, ep->ip, ep->port,
                global_kvraftclient_config_.GetConnectTimeoutMS(), nullptr, 0,
                *(global_kvraftclient_monitor_.get()))};
        if (open_ret) {
            socket.SetTimeout(global_kvraftclient_config_.GetSocketTimeoutMS());
            phxrpc::HttpMessageHandlerFactory http_msg_factory;
            KVRaftStub stub(socket, *(global_kvraftclient_monitor_.get()), http_msg_factory);
            return stub.RequestVote(req, resp);
        }

    }

    return -1;
}

int KVRaftClient::AppendEntries(const kvraft::AppendEntriesArgs &req, kvraft::AppendEntriesReply *resp)
{
    const phxrpc::Endpoint_t *ep{global_kvraftclient_config_.GetRandom()};

    if (ep) {
        phxrpc::BlockTcpStream socket;
        bool open_ret{phxrpc::PhxrpcTcpUtils::Open(&socket, ep->ip, ep->port,
                global_kvraftclient_config_.GetConnectTimeoutMS(), nullptr, 0,
                *(global_kvraftclient_monitor_.get()))};
        if (open_ret) {
            socket.SetTimeout(global_kvraftclient_config_.GetSocketTimeoutMS());
            phxrpc::HttpMessageHandlerFactory http_msg_factory;
            KVRaftStub stub(socket, *(global_kvraftclient_monitor_.get()), http_msg_factory);
            return stub.AppendEntries(req, resp);
        }

    }

    return -1;
}

int KVRaftClient::Command(const kvraft::KVArgs &req, kvraft::KVReply *resp)
{
    const phxrpc::Endpoint_t *ep{global_kvraftclient_config_.GetByIndex(0)};

    if (ep) {
        phxrpc::BlockTcpStream socket;
        bool open_ret{phxrpc::PhxrpcTcpUtils::Open(&socket, ep->ip, ep->port,
                global_kvraftclient_config_.GetConnectTimeoutMS(), nullptr, 0,
                *(global_kvraftclient_monitor_.get()))};
        if (open_ret) {
            socket.SetTimeout(global_kvraftclient_config_.GetSocketTimeoutMS());
            phxrpc::HttpMessageHandlerFactory http_msg_factory;
            KVRaftStub stub(socket, *(global_kvraftclient_monitor_.get()), http_msg_factory);
            return stub.Command(req, resp);
        }

    }

    return -1;
}