<commit_before>// Copyright 2017 <NAME>
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "base/logging.h"
#include "rpc/pb/raft_server.pb.h"

#include <sofa/pbrpc/pbrpc.h>

namespace consensus {
namespace rpc {

class SyncRaftClient : sofa::pbrpc::RpcClient {
 public:
  explicit SyncRaftClient(const std::string& url)
      : channel_(this, url, sofa::pbrpc::RpcChannelOptions()) {}

  // Synchronously sending request to specified url.
  // NOTE: The ownership of `msg` will be taken over by this method.
  StatusWith<pb::Response> Step(yaraft::pb::Message* msg) {
    sofa::pbrpc::RpcController* cntl = new sofa::pbrpc::RpcController();
    cntl->SetTimeout(3000);

    rpc::pb::Response response;
    rpc::pb::Request request;
    request.set_allocated_message(msg);

    rpc::pb::RaftService_Stub stub(&channel_);
    stub.Step(cntl, &request, &response, NULL);

    if (cntl->Failed()) {
      auto errStr = fmt::format("request failed: {}", cntl->ErrorText());
      LOG(ERROR) << errStr;
      return Status::Make(Error::RpcError, errStr);
    } else {
      FMT_SLOG(INFO, "request succeed with response: {%s}", response.ShortDebugString());
    }

    delete cntl;
    return response;
  }

 private:
  sofa::pbrpc::RpcChannel channel_;
};

class AsyncRaftClient : sofa::pbrpc::RpcClient {
 public:
  explicit AsyncRaftClient(const std::string& url)
      : channel_(this, url, sofa::pbrpc::RpcChannelOptions()) {}

  // Asynchronously sending request to specified url.
  void Step(yaraft::pb::Message* msg) {
    // -- prepare parameters --

    auto cntl = new sofa::pbrpc::RpcController();
    cntl->SetTimeout(3000);

    auto request = new rpc::pb::Request();
    request->set_allocated_message(msg);

    auto response = new rpc::pb::Response();

    auto done =
        sofa::pbrpc::NewClosure(this, &AsyncRaftClient::doneCallBack, cntl, request, response);

    // -- request --

    rpc::pb::RaftService_Stub stub(&channel_);
    stub.Step(cntl, request, response, done);
  }

  // `onSuccess` should not destroy the response object.
  void RegisterOnSuccess(std::function<void(rpc::pb::Response*)> onSuccess) {
    onSuccess_ = onSuccess;
  }

  void RegisterOnFail(std::function<void()> onFail) {
    onFail_ = onFail;
  }

 private:
  void doneCallBack(const sofa::pbrpc::RpcController* cntl, rpc::pb::Request* request,
                    rpc::pb::Response* response) {
    if (cntl->Failed()) {
      FMT_SLOG(ERROR, "request failed: %s", cntl->ErrorText().c_str());

      if (onFail_) {
        onFail_();
      }
    } else {
      FMT_SLOG(INFO, "request succeed with response: {%s}", response->ShortDebugString());

      if (onSuccess_) {
        onSuccess_(response);
      }
      delete response;
    }
    delete cntl;
    delete request;
  }

 private:
  sofa::pbrpc::RpcChannel channel_;
  std::function<void(rpc::pb::Response*)> onSuccess_;
  std::function<void()> onFail_;
};

}  // namespace rpc
}  // namespace consensus<commit_msg>rpc/raft_client: refactor as the raft_server.proto changed.<commit_after>// Copyright 2017 <NAME>
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#pragma once

#include "base/logging.h"
#include "rpc/pb/raft_server.pb.h"

#include <sofa/pbrpc/pbrpc.h>

namespace consensus {
namespace rpc {

class SyncRaftClient : sofa::pbrpc::RpcClient {
 public:
  explicit SyncRaftClient(const std::string& url)
      : channel_(this, url, sofa::pbrpc::RpcChannelOptions()) {}

  // Synchronously sending request to specified url.
  // NOTE: The ownership of `msg` will be taken over by this method.
  StatusWith<pb::Response> Step(yaraft::pb::Message* msg) {
    rpc::pb::Request request;
    request.set_allocated_message(msg);

    return sendRequest(request);
  }

  StatusWith<pb::Response> Tick(uint32_t ticks) {
    rpc::pb::Request request;
    request.set_ticks(ticks);

    return sendRequest(request);
  }

 private:
  StatusWith<pb::Response> sendRequest(rpc::pb::Request& request) {
    sofa::pbrpc::RpcController cntl;
    cntl.SetTimeout(3000);

    rpc::pb::Response response;
    rpc::pb::RaftService_Stub stub(&channel_);
    stub.Serve(&cntl, &request, &response, NULL);

    if (cntl.Failed()) {
      auto errStr = fmt::format("request failed: {}", cntl.ErrorText());
      LOG(ERROR) << errStr;
      return Status::Make(Error::RpcError, errStr);
    }
    return response;
  }

 private:
  sofa::pbrpc::RpcChannel channel_;
};

class AsyncRaftClient : sofa::pbrpc::RpcClient {
 public:
  explicit AsyncRaftClient(const std::string& url)
      : channel_(this, url, sofa::pbrpc::RpcChannelOptions()) {}

  // Asynchronously sending request to specified url.
  void Step(yaraft::pb::Message* msg) {
    // -- prepare parameters --

    auto cntl = new sofa::pbrpc::RpcController();
    cntl->SetTimeout(3000);

    auto request = new rpc::pb::Request();
    request->set_allocated_message(msg);

    auto response = new rpc::pb::Response();

    auto done =
        sofa::pbrpc::NewClosure(this, &AsyncRaftClient::doneCallBack, cntl, request, response);

    // -- request --

    rpc::pb::RaftService_Stub stub(&channel_);
    stub.Serve(cntl, request, response, done);
  }

  // `onSuccess` should not destroy the response object.
  void RegisterOnSuccess(std::function<void(rpc::pb::Response*)> onSuccess) {
    onSuccess_ = onSuccess;
  }

  void RegisterOnFail(std::function<void()> onFail) {
    onFail_ = onFail;
  }

 private:
  void doneCallBack(const sofa::pbrpc::RpcController* cntl, rpc::pb::Request* request,
                    rpc::pb::Response* response) {
    if (cntl->Failed()) {
      FMT_SLOG(ERROR, "request failed: %s", cntl->ErrorText().c_str());

      if (onFail_) {
        onFail_();
      }
    } else {
      if (onSuccess_) {
        onSuccess_(response);
      }
      delete response;
    }
    delete cntl;
    delete request;
  }

 private:
  sofa::pbrpc::RpcChannel channel_;
  std::function<void(rpc::pb::Response*)> onSuccess_;
  std::function<void()> onFail_;
};

}  // namespace rpc
}  // namespace consensus
{'fraction_non_alphanumeric': 0.13998299801643524, 'fraction_numerical': 0.004817228676678946, 'mean_word_length': 3.6532630191166775, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2933625', 'n_tokens_mistral': 2267, 'n_tokens_neox': 2222, 'n_words': 563}