Merge branch 'master' into scan-range-fix

Author: Hoverbear

Cargo.toml

@@ -21,10 +21,10 @@ name = "tikv_client"
 failure = "0.1"
 futures = "0.1"
 fxhash = "0.2"
-grpcio = { version = "0.4", features = [ "secure" ] }
+grpcio = { version = "0.5.0-alpha", features = [ "secure", "prost-codec" ], default-features = false }
 lazy_static = "0.2.1"
 log = "0.3.9"
-protobuf = "~2.0"
+protobuf = "2"
 serde = "1.0"
 serde_derive = "1.0"
 tokio-core = "0.1"

src/errors.rs

@@ -22,6 +22,7 @@ pub struct Error {
 
 /// An error originating from the TiKV client or dependencies.
 #[derive(Debug, Fail)]
+#[allow(clippy::large_enum_variant)]
 pub enum ErrorKind {
     /// Wraps a `std::io::Error`.
     #[fail(display = "IO error: {}", _0)]

src/rpc/client.rs

@@ -29,6 +29,7 @@ use futures::{
 use grpcio::{EnvBuilder, Environment};
 use kvproto::kvrpcpb;
 use log::*;
+use protobuf::Message;
 
 use crate::{
     raw::ColumnFamily,

src/rpc/pd/client.rs

@@ -19,7 +19,7 @@ use std::{
 
 use futures::Future;
 use grpcio::{CallOption, Environment};
-use kvproto::{metapb, pdpb, pdpb_grpc::PdClient as RpcClient};
+use kvproto::{metapb, pdpb, pdpb::PdClient as RpcClient};
 
 use crate::{
     rpc::{
@@ -117,7 +117,7 @@ impl PdClient {
         &self,
         region_id: u64,
     ) -> impl Future<Item = (metapb::Region, Option<metapb::Peer>), Error = Error> {
-        let mut req = pd_request!(self.cluster_id, pdpb::GetRegionByIDRequest);
+        let mut req = pd_request!(self.cluster_id, pdpb::GetRegionByIdRequest);
         req.set_region_id(region_id);
 
         self.execute(request_context(

src/rpc/pd/leader.rs

@@ -26,8 +26,9 @@ use futures::{
 };
 use fxhash::FxHashSet as HashSet;
 use grpcio::{CallOption, Environment, WriteFlags};
-use kvproto::{pdpb, pdpb_grpc};
+use kvproto::pdpb;
 use log::*;
+use protobuf::Message;
 use tokio_core::reactor::{Core, Handle as OtherHandle};
 
 use crate::{
@@ -44,6 +45,7 @@ use crate::{
 
 macro_rules! pd_request {
     ($cluster_id:expr, $type:ty) => {{
+        use ::protobuf::Message;
         let mut request = <$type>::new();
         let mut header = ::kvproto::pdpb::RequestHeader::new();
         header.set_cluster_id($cluster_id);
@@ -225,7 +227,7 @@ impl PdReactor {
 }
 
 pub struct LeaderClient {
-    pub client: pdpb_grpc::PdClient,
+    pub client: pdpb::PdClient,
     pub members: pdpb::GetMembersResponse,
 
     env: Arc<Environment>,
@@ -304,7 +306,7 @@ pub fn validate_endpoints(
     endpoints: &[String],
     security_mgr: &SecurityManager,
     timeout: Duration,
-) -> Result<(pdpb_grpc::PdClient, pdpb::GetMembersResponse)> {
+) -> Result<(pdpb::PdClient, pdpb::GetMembersResponse)> {
     let len = endpoints.len();
     let mut endpoints_set = HashSet::with_capacity_and_hasher(len, Default::default());
 
@@ -359,8 +361,8 @@ fn connect(
     security_mgr: &SecurityManager,
     addr: &str,
     timeout: Duration,
-) -> Result<(pdpb_grpc::PdClient, pdpb::GetMembersResponse)> {
-    let client = security_mgr.connect(env, addr, pdpb_grpc::PdClient::new)?;
+) -> Result<(pdpb::PdClient, pdpb::GetMembersResponse)> {
+    let client = security_mgr.connect(env, addr, pdpb::PdClient::new)?;
     let option = CallOption::default().timeout(timeout);
     let resp = client
         .get_members_opt(&pdpb::GetMembersRequest::new(), option)
@@ -374,7 +376,7 @@ fn try_connect(
     addr: &str,
     cluster_id: u64,
     timeout: Duration,
-) -> Result<(pdpb_grpc::PdClient, pdpb::GetMembersResponse)> {
+) -> Result<(pdpb::PdClient, pdpb::GetMembersResponse)> {
     let (client, r) = connect(Arc::clone(&env), security_mgr, addr, timeout)?;
     let new_cluster_id = r.get_header().get_cluster_id();
     if new_cluster_id != cluster_id {
@@ -394,7 +396,7 @@ pub fn try_connect_leader(
     security_mgr: &SecurityManager,
     previous: &pdpb::GetMembersResponse,
     timeout: Duration,
-) -> Result<(pdpb_grpc::PdClient, pdpb::GetMembersResponse)> {
+) -> Result<(pdpb::PdClient, pdpb::GetMembersResponse)> {
     let previous_leader = previous.get_leader();
     let members = previous.get_members();
     let cluster_id = previous.get_header().get_cluster_id();

src/rpc/tikv/client.rs

@@ -18,8 +18,8 @@ use std::{fmt, sync::Arc, time::Duration};
 
 use futures::Future;
 use grpcio::{CallOption, Environment};
-use kvproto::{errorpb, kvrpcpb, tikvpb_grpc::TikvClient};
-use protobuf;
+use kvproto::{errorpb, kvrpcpb, tikvpb::TikvClient};
+use protobuf::{self, Message};
 
 use crate::{
     rpc::{
@@ -95,7 +95,7 @@ has_region_error!(kvrpcpb::CleanupResponse);
 has_region_error!(kvrpcpb::BatchGetResponse);
 has_region_error!(kvrpcpb::ScanLockResponse);
 has_region_error!(kvrpcpb::ResolveLockResponse);
-has_region_error!(kvrpcpb::GCResponse);
+has_region_error!(kvrpcpb::GcResponse);
 has_region_error!(kvrpcpb::RawGetResponse);
 has_region_error!(kvrpcpb::RawBatchGetResponse);
 has_region_error!(kvrpcpb::RawPutResponse);
@@ -127,7 +127,7 @@ has_key_error!(kvrpcpb::BatchRollbackResponse);
 has_key_error!(kvrpcpb::CleanupResponse);
 has_key_error!(kvrpcpb::ScanLockResponse);
 has_key_error!(kvrpcpb::ResolveLockResponse);
-has_key_error!(kvrpcpb::GCResponse);
+has_key_error!(kvrpcpb::GcResponse);
 
 macro_rules! has_str_error {
     ($type:ty) => {
@@ -426,8 +426,8 @@ impl KvClient {
         &self,
         context: TxnContext,
         safe_point: u64,
-    ) -> impl Future<Item = kvrpcpb::GCResponse, Error = Error> {
-        let mut req = txn_request!(context, kvrpcpb::GCRequest);
+    ) -> impl Future<Item = kvrpcpb::GcResponse, Error = Error> {
+        let mut req = txn_request!(context, kvrpcpb::GcRequest);
         req.set_safe_point(safe_point);
 
         self.execute(request_context(
@@ -643,7 +643,7 @@ impl KvClient {
     }
 
     #[inline]
-    fn convert_to_grpc_pairs(pairs: Vec<KvPair>) -> protobuf::RepeatedField<kvrpcpb::KvPair> {
+    fn convert_to_grpc_pairs(pairs: Vec<KvPair>) -> Vec<kvrpcpb::KvPair> {
         pairs.into_iter().map(Self::convert_to_grpc_pair).collect()
     }
 
@@ -653,9 +653,8 @@ impl KvClient {
     }
 
     #[inline]
-    fn convert_from_grpc_pairs(pairs: protobuf::RepeatedField<kvrpcpb::KvPair>) -> Vec<KvPair> {
+    fn convert_from_grpc_pairs(pairs: Vec<kvrpcpb::KvPair>) -> Vec<KvPair> {
         pairs
-            .into_vec()
             .into_iter()
             .map(Self::convert_from_grpc_pair)
             .collect()
@@ -673,7 +672,7 @@ impl KvClient {
     #[inline]
     fn convert_to_grpc_ranges(
         ranges: impl Iterator<Item = (Option<Key>, Option<Key>)>,
-    ) -> protobuf::RepeatedField<kvrpcpb::KeyRange> {
+    ) -> Vec<kvrpcpb::KeyRange> {
         ranges.map(Self::convert_to_grpc_range).collect()
     }
 }