Remove compat

Cargo.toml

@@ -20,11 +20,11 @@ name = "tikv_client"
 [dependencies]
 derive-new = "0.5"
 failure = "0.1"
-futures = { version = "0.3.1", features = ["compat", "async-await", "thread-pool"] }
-grpcio = { version = "0.5.0", features = [ "secure", "prost-codec" ], default-features = false }
-kvproto = { git = "https://github.com/pingcap/kvproto.git", features = [ "prost-codec" ], default-features = false }
+futures = { version = "0.3.5", features = ["async-await", "thread-pool"] }
+grpcio = { version = "0.6", features = [ "secure", "prost-codec" ], default-features = false }
+kvproto = { git = "https://github.com/pingcap/kvproto.git", rev = "1e28226154c374788f38d3a542fc505cd74720f3", features = [ "prost-codec" ], default-features = false }
 lazy_static = "1"
-log = "0.4"
+log = "0.4" 
 rand = "0.7"
 regex = "1"
 serde = "1.0"

src/kv_client/mod.rs

@@ -0,0 +1,127 @@
+// Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
+
+mod errors;
+
+pub use self::errors::{HasError, HasRegionError};
+pub use kvproto::tikvpb::TikvClient;
+
+use crate::{
+    pd::Region,
+    request::{KvRequest, KvRpcRequest},
+    security::SecurityManager,
+    stats::tikv_stats,
+    ErrorKind, Result,
+};
+
+use derive_new::new;
+use futures::future::BoxFuture;
+use futures::prelude::*;
+use grpcio::CallOption;
+use grpcio::Environment;
+use std::sync::Arc;
+use std::time::Duration;
+
+/// A trait for connecting to TiKV stores.
+pub trait KvConnect: Sized {
+    type KvClient: KvClient + Clone + Send + Sync + 'static;
+
+    fn connect(&self, address: &str) -> Result<Self::KvClient>;
+}
+
+pub type RpcFnType<Req, Resp> =
+    for<'a, 'b> fn(
+        &'a TikvClient,
+        &'b Req,
+        CallOption,
+    )
+        -> std::result::Result<::grpcio::ClientUnaryReceiver<Resp>, ::grpcio::Error>;
+
+#[derive(new, Clone)]
+pub struct TikvConnect {
+    env: Arc<Environment>,
+    security_mgr: Arc<SecurityManager>,
+}
+
+impl KvConnect for TikvConnect {
+    type KvClient = KvRpcClient;
+
+    fn connect(&self, address: &str) -> Result<KvRpcClient> {
+        self.security_mgr
+            .connect(self.env.clone(), address, TikvClient::new)
+            .map(|c| KvRpcClient::new(Arc::new(c)))
+    }
+}
+
+pub trait KvClient {
+    fn dispatch<T: KvRequest>(
+        &self,
+        request: &T,
+        opt: CallOption,
+    ) -> BoxFuture<'static, Result<T::RpcResponse>>;
+}
+
+/// This client handles requests for a single TiKV node. It converts the data
+/// types and abstractions of the client program into the grpc data types.
+#[derive(new, Clone)]
+pub struct KvRpcClient {
+    rpc_client: Arc<TikvClient>,
+}
+
+impl KvClient for KvRpcClient {
+    fn dispatch<T: KvRequest>(
+        &self,
+        request: &T,
+        opt: CallOption,
+    ) -> BoxFuture<'static, Result<T::RpcResponse>> {
+        map_errors_and_trace(T::REQUEST_NAME, T::RPC_FN(&self.rpc_client, request, opt)).boxed()
+    }
+}
+
+async fn map_errors_and_trace<Resp, RpcFuture>(
+    request_name: &'static str,
+    fut: ::grpcio::Result<RpcFuture>,
+) -> Result<Resp>
+where
+    RpcFuture: Future<Output = std::result::Result<Resp, ::grpcio::Error>>,
+    Resp: HasError + Sized + Clone + Send + 'static,
+{
+    let res = match fut {
+        Ok(f) => f.await,
+        Err(e) => Err(e),
+    };
+
+    let context = tikv_stats(request_name);
+    context.done(res.map_err(|e| ErrorKind::Grpc(e).into()))
+}
+
+#[derive(new)]
+pub struct Store<Client: KvClient> {
+    pub region: Region,
+    client: Client,
+    timeout: Duration,
+}
+
+impl<Client: KvClient> Store<Client> {
+    pub fn call_options(&self) -> CallOption {
+        CallOption::default().timeout(self.timeout)
+    }
+
+    pub fn request<T: KvRpcRequest>(&self) -> T {
+        let mut request = T::default();
+        // FIXME propagate the error instead of using `expect`
+        request.set_context(
+            self.region
+                .context()
+                .expect("Cannot create context from region"),
+        );
+        request
+    }
+
+    pub fn dispatch<T: KvRequest>(
+        &self,
+        request: &T,
+        opt: CallOption,
+    ) -> BoxFuture<'static, Result<T::RpcResponse>> {
+        self.client.dispatch::<T>(request, opt)
+    }
+}

src/mock.rs

@@ -11,10 +11,7 @@ use crate::{
     Config, Error, Key, Result, Timestamp,
 };
 use fail::fail_point;
-use futures::{
-    compat::Compat01As03,
-    future::{ready, BoxFuture, FutureExt},
-};
+use futures::future::{ready, BoxFuture, FutureExt};
 use grpcio::CallOption;
 use kvproto::{errorpb, kvrpcpb, metapb, tikvpb::TikvClient};
 use std::{future::Future, sync::Arc, time::Duration};
@@ -59,7 +56,7 @@ impl KvClient for MockKvClient {
         _fut: grpcio::Result<RpcFuture>,
     ) -> BoxFuture<'static, Result<Resp>>
     where
-        Compat01As03<RpcFuture>: Future<Output = std::result::Result<Resp, ::grpcio::Error>>,
+        RpcFuture: Future<Output = std::result::Result<Resp, ::grpcio::Error>>,
         Resp: HasError + Sized + Clone + Send + 'static,
         RpcFuture: Send + 'static,
     {

tikv-client-common/Cargo.toml

@@ -6,8 +6,8 @@ edition = "2018"
 
 [dependencies]
 failure = "0.1"
-grpcio = { version = "0.5.0", features = [ "secure", "prost-codec" ], default-features = false }
-futures = { version = "0.3.1", features = ["compat", "async-await", "thread-pool"] }
+grpcio = { version = "0.6", features = [ "secure", "prost-codec" ], default-features = false }
+futures = { version = "0.3.5", features = ["compat", "async-await", "thread-pool"] }
 lazy_static = "1"
 regex = "1"
 serde = "1.0"
@@ -16,7 +16,7 @@ log = "0.4"
 proptest = "0.9"
 proptest-derive = "0.1.0"
 derive-new = "0.5"
-kvproto = { git = "https://github.com/pingcap/kvproto.git", features = [ "prost-codec" ], default-features = false }
+kvproto = { git = "https://github.com/pingcap/kvproto.git", rev = "1e28226154c374788f38d3a542fc505cd74720f3", features = [ "prost-codec" ], default-features = false }
 
 [dependencies.prometheus]
 version = "0.8"

tikv-client-pd/Cargo.toml

@@ -6,10 +6,10 @@ edition = "2018"
 
 [dependencies]
 derive-new = "0.5"
-kvproto = { git = "https://github.com/pingcap/kvproto.git", features = [ "prost-codec" ], default-features = false }
-futures = { version = "0.3.1", features = ["compat", "async-await", "thread-pool"] }
+kvproto = { git = "https://github.com/pingcap/kvproto.git", rev = "1e28226154c374788f38d3a542fc505cd74720f3", features = [ "prost-codec" ], default-features = false }
+futures = { version = "0.3.5", features = ["compat", "async-await", "thread-pool"] }
 tokio = { version = "0.2", features = ["sync"] }
-grpcio = { version = "0.5.0", features = [ "secure", "prost-codec" ], default-features = false }
+grpcio = { version = "0.6", features = [ "secure", "prost-codec" ], default-features = false }
 log = "0.4"
 
 tikv-client-common = { path = "../tikv-client-common" }

tikv-client-pd/src/cluster.rs

@@ -4,7 +4,7 @@
 #![allow(dead_code)]
 
 use crate::timestamp::TimestampOracle;
-use futures::{compat::Compat01As03, prelude::*};
+use futures::prelude::*;
 use grpcio::{CallOption, Environment};
 use kvproto::{metapb, pdpb};
 use std::{
@@ -51,7 +51,6 @@ impl Cluster {
 
         self.client
             .get_region_async_opt(&req, option)
-            .map(Compat01As03::new)
             .unwrap()
             .map(move |r| context.done(r.map_err(|e| e.into())))
             .and_then(move |resp| {
@@ -82,7 +81,6 @@ impl Cluster {
 
         self.client
             .get_region_by_id_async_opt(&req, option)
-            .map(Compat01As03::new)
             .unwrap()
             .map(move |r| context.done(r.map_err(|e| e.into())))
             .and_then(move |resp| {
@@ -111,7 +109,6 @@ impl Cluster {
 
         self.client
             .get_store_async_opt(&req, option)
-            .map(Compat01As03::new)
             .unwrap()
             .map(move |r| context.done(r.map_err(|e| e.into())))
             .and_then(|mut resp| {
@@ -136,7 +133,6 @@ impl Cluster {
 
         self.client
             .get_all_stores_async_opt(&req, option)
-            .map(Compat01As03::new)
             .unwrap()
             .map(move |r| context.done(r.map_err(|e| e.into())))
             .and_then(|mut resp| {
@@ -275,7 +271,6 @@ impl Connection {
         let option = CallOption::default().timeout(timeout);
         let resp = client
             .get_members_async_opt(&pdpb::GetMembersRequest::default(), option)
-            .map(Compat01As03::new)
             .map_err(Error::from)?
             .await?;
         Ok((client, resp))

tikv-client-pd/src/timestamp.rs

@@ -13,7 +13,6 @@
 
 use futures::{
     channel::{mpsc, oneshot},
-    compat::*,
     executor::block_on,
     join, pin_mut,
     prelude::*,
@@ -53,8 +52,8 @@ impl TimestampOracle {
         thread::spawn(move || {
             block_on(run_tso(
                 cluster_id,
-                rpc_sender.sink_compat().sink_err_into(),
-                rpc_receiver.compat().err_into(),
+                rpc_sender.sink_err_into(),
+                rpc_receiver.err_into(),
                 request_rx,
             ))
         });

tikv-client-store/Cargo.toml

@@ -5,10 +5,10 @@ edition = "2018"
 
 
 [dependencies]
-futures = { version = "0.3.1", features = ["compat", "async-await", "thread-pool"] }
-kvproto = { git = "https://github.com/pingcap/kvproto.git", features = [ "prost-codec" ], default-features = false }
+futures = { version = "0.3.5", features = ["compat", "async-await", "thread-pool"] }
+kvproto = { git = "https://github.com/pingcap/kvproto.git", rev = "1e28226154c374788f38d3a542fc505cd74720f3", features = [ "prost-codec" ], default-features = false }
 derive-new = "0.5"
-grpcio = { version = "0.5.0", features = [ "secure", "prost-codec" ], default-features = false }
+grpcio = { version = "0.6", features = [ "secure", "prost-codec" ], default-features = false }
 log = "0.4"
 
 tikv-client-common = { path = "../tikv-client-common" }

tikv-client-store/src/lib.rs

@@ -7,7 +7,7 @@ mod errors;
 
 pub use self::errors::{HasError, HasRegionError};
 use derive_new::new;
-use futures::{compat::Compat01As03, future::BoxFuture, prelude::*};
+use futures::{future::BoxFuture, prelude::*};
 use grpcio::{CallOption, Environment};
 pub use kvproto::tikvpb::TikvClient;
 use std::{sync::Arc, time::Duration};
@@ -53,7 +53,7 @@ pub trait KvClient {
         fut: ::grpcio::Result<RpcFuture>,
     ) -> BoxFuture<'static, Result<Resp>>
     where
-        Compat01As03<RpcFuture>: Future<Output = std::result::Result<Resp, ::grpcio::Error>>,
+        RpcFuture: Future<Output = std::result::Result<Resp, ::grpcio::Error>>,
         Resp: HasError + Sized + Clone + Send + 'static,
         RpcFuture: Send + 'static;
 
@@ -74,7 +74,7 @@ impl KvClient for KvRpcClient {
         fut: ::grpcio::Result<RpcFuture>,
     ) -> BoxFuture<'static, Result<Resp>>
     where
-        Compat01As03<RpcFuture>: Future<Output = std::result::Result<Resp, ::grpcio::Error>>,
+        RpcFuture: Future<Output = std::result::Result<Resp, ::grpcio::Error>>,
         Resp: HasError + Sized + Clone + Send + 'static,
         RpcFuture: Send + 'static,
     {
@@ -114,7 +114,7 @@ impl<Client: KvClient> Store<Client> {
         fut: ::grpcio::Result<RpcFuture>,
     ) -> BoxFuture<'static, Result<Resp>>
     where
-        Compat01As03<RpcFuture>: Future<Output = std::result::Result<Resp, ::grpcio::Error>>,
+        RpcFuture: Future<Output = std::result::Result<Resp, ::grpcio::Error>>,
         Resp: HasError + Sized + Clone + Send + 'static,
         RpcFuture: Send + 'static,
     {
@@ -127,11 +127,11 @@ async fn map_errors_and_trace<Resp, RpcFuture>(
     fut: ::grpcio::Result<RpcFuture>,
 ) -> Result<Resp>
 where
-    Compat01As03<RpcFuture>: Future<Output = std::result::Result<Resp, ::grpcio::Error>>,
+    RpcFuture: Future<Output = std::result::Result<Resp, ::grpcio::Error>>,
     Resp: HasError + Sized + Clone + Send + 'static,
 {
     let res = match fut {
-        Ok(f) => Compat01As03::new(f).await,
+        Ok(f) => f.await,
         Err(e) => Err(e),
     };