Delta V Software Forums

Code: Select all

class RpcClient {
 public:
  // Construct a RPC client for the RPC service designated by
  // "service_descriptor". To get a service descriptor, you can just use static
  // member function "descriptor()" of the corresponding service class.
  RpcClient(const google::protobuf::ServiceDescriptor& service_descriptor,
            const std::string& server_ip_list,
            const int port,
            const int timeout_ms);

  // Similar to above, but use servers listed in "FLAGS_rpc_server_ip_list",
  // "FLAGS_rpc_port" and "FLAGS_rpc_call_timeout_ms".
  RpcClient(const google::protobuf::ServiceDescriptor& service_descriptor);

  ~RpcClient();

  // Call the given method designated by "method_index" of the RPC service.
  // The method index is zero based. It is determined by the order that the
  // method was defined in proto file.
  // Return true if response is received successfully, otherwise return false.
  bool Call(const int method_index,
            const google::protobuf::Message& request,
            google::protobuf::Message* response);

  // Similar to above, just a simplified version for most scenarios.
  bool Call(const google::protobuf::Message& request,
            google::protobuf::Message* response) {
    return Call(0, request, response);
  }

 private:
  google::protobuf::RpcChannel* AllocateChannel();
  void ReclaimChannel(google::protobuf::RpcChannel* channel);

  const RCF::TcpEndpoint& GetServer();

  const google::protobuf::ServiceDescriptor& service_descriptor_;

  const int timeout_ms_;

  std::mutex mutex_;
  size_t next_server_index_ = 0;                                                                                                                                                                                                                                              
  std::vector<RCF::TcpEndpoint> servers_;
  std::queue<google::protobuf::RpcChannel*> channels_;
};

Code: Select all

namespace {

class InternalRpcChannel : public RCF::RcfProtoChannel {
 public:
  InternalRpcChannel(const RCF::TcpEndpoint& endpoint)
      : RCF::RcfProtoChannel(endpoint), endpoint_(endpoint) {
  }

  const RCF::TcpEndpoint& endpoint() const {
    return endpoint_;
  }

 private:
  // We use reference here, becuase we know it is not a dangling one.
  const RCF::TcpEndpoint& endpoint_;
};

} // namespace

RpcClient::RpcClient(const ServiceDescriptor& service_descriptor,
                     const std::string& server_ip_list,
                     const int port,
                     const int timeout_ms)
    : service_descriptor_(service_descriptor), timeout_ms_(timeout_ms) {
  RCF::init();

  std::vector<std::string> ips;
  SplitStringUsing(server_ip_list, ",", &ips);
  for (const std::string& ip : ips) {
    servers_.emplace_back(RCF::TcpEndpoint(ip, port));
  }

  next_server_index_ = std::rand() % servers_.size();
}

RpcClient::RpcClient(const ServiceDescriptor& service_descriptor)
  : RpcClient(service_descriptor,
              FLAGS_rpc_server_ip_list,
              FLAGS_rpc_port,
              FLAGS_rpc_call_timeout_ms) {
}

RpcClient::~RpcClient() {
  LOG(INFO) << "Destructing RpcClient.";
  while (!channels_.empty()) {
    delete channels_.front();
    channels_.pop();
  }
}

bool RpcClient::Call(const int method_index,
                     const google::protobuf::Message& request,
                     google::protobuf::Message* response) {
  google::protobuf::RpcChannel* channel = AllocateChannel();
  try {
    channel->CallMethod(service_descriptor_.method(method_index),
                        nullptr,
                        &request,
                        response,
                        nullptr);
  } catch (const RCF::Exception& exception) {
    LOG(ERROR) << service_descriptor_.full_name() << " RPC exception: " <<
        exception.getErrorString() << " Server: " <<
        static_cast<InternalRpcChannel*>(channel)->endpoint().getIp();
    delete channel;
    return false;
  }

  ReclaimChannel(channel);
  return true;
}

google::protobuf::RpcChannel* RpcClient::AllocateChannel() {
  std::lock_guard<std::mutex> guard(mutex_);
  if (channels_.empty()) {
    const RCF::TcpEndpoint& endpoint = GetServer();
    LOG(INFO) << "New " << service_descriptor_.full_name() <<
        " RpcChannel instance to " << endpoint.asString();
    RCF::RcfProtoChannel* channel = new InternalRpcChannel(endpoint);
    channel->setRemoteCallTimeoutMs(timeout_ms_);
    return channel;
  }

  google::protobuf::RpcChannel* channel = channels_.front();
  channels_.pop();

  return channel;
}

inline void RpcClient::ReclaimChannel(google::protobuf::RpcChannel* channel) {
  std::lock_guard<std::mutex> guard(mutex_);
  channels_.push(channel);
}

const RCF::TcpEndpoint& RpcClient::GetServer() {
  if (next_server_index_ == servers_.size()) {
    next_server_index_ = 0;
    return servers_[next_server_index_];
  } else {
    return servers_[next_server_index_++];
  }
}