Merge branch 'master' into mods

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

@@ -12,6 +12,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

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

src/rpc/pd/client.rs

@@ -8,7 +8,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::{
@@ -106,7 +106,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

@@ -15,8 +15,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::{
@@ -33,6 +34,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);
@@ -214,7 +216,7 @@ impl PdReactor {
 }
 
 pub struct LeaderClient {
-    pub client: pdpb_grpc::PdClient,
+    pub client: pdpb::PdClient,
     pub members: pdpb::GetMembersResponse,
 
     env: Arc<Environment>,
@@ -293,7 +295,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());
 
@@ -348,8 +350,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)
@@ -363,7 +365,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 {
@@ -383,7 +385,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

@@ -7,8 +7,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::{
@@ -84,7 +84,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);
@@ -116,7 +116,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) => {
@@ -415,8 +415,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(
@@ -632,7 +632,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()
     }
 
@@ -642,9 +642,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()
@@ -662,7 +661,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()
     }
 }