Address review comments

Signed-off-by: Nick Cameron <[email protected]>

Author: nrc

.travis.yml

@@ -6,14 +6,14 @@ os:
   # - windows # TODO: https://github.com/pingcap/kvproto/issues/355
   - osx
 rust:
-  # - stable
+  - stable
   - nightly
 
 env:
   global:
     - RUST_BACKTRACE=1
-    - RUSTFLAGS="-D warnings"
-    - RUSTDOCFLAGS="-D warnings" 
+#    - RUSTFLAGS="-D warnings"
+#    - RUSTDOCFLAGS="-D warnings" 
 
 addons:
   apt:
@@ -43,7 +43,7 @@ script:
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "stable" ]]; then cargo clippy -- -D clippy::all; fi
   - cargo build --all
   - cargo build --examples
-  - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then cargo doc --exclude tikv-client-proto --document-private-items; fi
+  - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then cargo doc --workspace --exclude tikv-client-proto --document-private-items --no-deps; fi
   - if [[ $TRAVIS_RUST_VERSION == "nightly" ]]; then cargo test --all -- --nocapture; fi
   # For now we only run full integration tests on Linux. Here's why:
   # * Docker on OS X is not supported by Travis.
@@ -53,7 +53,7 @@ script:
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then docker ps; fi
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then docker logs pd; fi
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then docker logs kv; fi
-  - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then sleep 2; fi
+  - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then sleep 60; fi
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then PD_ADDRS="127.0.0.1:2379" cargo test --all --features integration-tests -- --nocapture; fi
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then cargo run --example raw -- --pd="127.0.0.1:2379"; fi
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then cargo run --example transaction -- --pd="127.0.0.1:2379"; fi

Cargo.toml

@@ -22,7 +22,7 @@ async-trait = "0.1"
 derive-new = "0.5"
 futures = { version = "0.3.5", features = ["async-await", "thread-pool"] }
 futures-timer = "3.0"
-grpcio = { version = "0.7", features = [ "secure", "prost-codec" ], default-features = false }
+grpcio = { version = "0.8", features = [ "secure", "prost-codec", "use-bindgen" ], default-features = false }
 lazy_static = "1"
 log = "0.4"
 prometheus = { version = "0.11", features = [ "push", "process" ], default-features = false } 

mock-tikv/Cargo.toml

@@ -6,6 +6,6 @@ edition = "2018"
 [dependencies]
 derive-new = "0.5.8"
 futures = "0.3.5"
-grpcio = { version = "0.7", features = [ "secure", "prost-codec" ], default-features = false }
+grpcio = { version = "0.8", features = [ "secure", "prost-codec", "use-bindgen" ], default-features = false }
 log = "0.4"
 tikv-client-proto = { path = "../tikv-client-proto"}

rustfmt.toml

@@ -1,2 +1,2 @@
-merge_imports = true
+imports_granularity="Crate"
 format_code_in_doc_comments = true

src/compat.rs

@@ -18,6 +18,7 @@ use std::pin::Pin;
 pub struct LoopFn<A, F> {
     future: A,
     func: F,
+    errored: bool,
 }
 
 pub fn stream_fn<S, T, A, F, E>(initial_state: S, mut func: F) -> LoopFn<A, F>
@@ -28,6 +29,7 @@ where
     LoopFn {
         future: func(initial_state),
         func,
+        errored: false,
     }
 }
 
@@ -38,10 +40,16 @@ where
 {
     type Item = Result<T, E>;
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        if self.errored {
+            return Poll::Ready(None);
+        }
         unsafe {
             let this = Pin::get_unchecked_mut(self);
             match ready!(Pin::new_unchecked(&mut this.future).poll(cx)) {
-                Err(e) => Poll::Ready(Some(Err(e))),
+                Err(e) => {
+                    this.errored = true;
+                    Poll::Ready(Some(Err(e)))
+                }
                 Ok(None) => Poll::Ready(None),
                 Ok(Some((s, t))) => {
                     this.future = (this.func)(s);

src/lib.rs

@@ -5,9 +5,6 @@
 #![allow(clippy::redundant_closure)]
 #![allow(clippy::type_complexity)]
 #![allow(incomplete_features)]
-#![cfg_attr(test, feature(specialization))]
-#![feature(associated_type_bounds)]
-#![feature(decl_macro)]
 
 //! This crate provides a clean, ready to use client for [TiKV](https://github.com/tikv/tikv), a
 //! distributed transactional Key-Value database written in Rust.
@@ -77,6 +74,9 @@
 //!
 //! At this point, you should seek the documentation in the related API modules.
 
+#[macro_use]
+mod request;
+
 #[macro_use]
 mod transaction;
 
@@ -87,7 +87,6 @@ mod kv;
 mod pd;
 mod raw;
 mod region;
-mod request;
 mod stats;
 mod store;
 mod timestamp;

src/mock.rs

@@ -154,3 +154,10 @@ impl PdClient for MockPdClient {
         unimplemented!()
     }
 }
+
+pub fn mock_store() -> Store {
+    Store {
+        region: Region::default(),
+        client: Arc::new(MockKvClient::new("foo".to_owned(), None)),
+    }
+}

src/raw/client.rs

@@ -117,7 +117,7 @@ impl Client {
             .await?
             .resolve_lock(OPTIMISTIC_BACKOFF)
             .retry_region(DEFAULT_REGION_BACKOFF)
-            .post_process()
+            .post_process_default()
             .plan();
         plan.execute().await
     }
@@ -147,8 +147,8 @@ impl Client {
         let request = new_raw_batch_get_request(keys.into_iter().map(Into::into), self.cf.clone());
         let plan = crate::request::PlanBuilder::new(self.rpc.clone(), request)
             .resolve_lock(OPTIMISTIC_BACKOFF)
-            .retry_region(DEFAULT_REGION_BACKOFF)
             .multi_region()
+            .retry_region(DEFAULT_REGION_BACKOFF)
             .merge(Collect)
             .plan();
         plan.execute()

src/raw/requests.rs

@@ -3,9 +3,7 @@
 use super::RawRpcRequest;
 use crate::{
     pd::PdClient,
-    request::{
-        shardable_keys, shardable_range, Collect, KvRequest, Merge, Process, Shardable, SingleKey,
-    },
+    request::{Collect, DefaultProcessor, KvRequest, Merge, Process, Shardable, SingleKey},
     store::{store_stream_for_keys, store_stream_for_ranges, Store},
     transaction::HasLocks,
     util::iter::FlatMapOkIterExt,
@@ -33,10 +31,10 @@ impl SingleKey for kvrpcpb::RawGetRequest {
     }
 }
 
-impl Process for kvrpcpb::RawGetResponse {
+impl Process<kvrpcpb::RawGetResponse> for DefaultProcessor {
     type Out = Option<Value>;
 
-    fn process(input: Result<Self>) -> Result<Self::Out> {
+    fn process(&self, input: Result<kvrpcpb::RawGetResponse>) -> Result<Self::Out> {
         let mut input = input?;
         Ok(if input.not_found {
             None

src/request/mod.rs

@@ -10,15 +10,16 @@ use tikv_client_store::{HasError, Request};
 
 pub use self::{
     plan::{
-        Collect, CollectError, Dispatch, Merge, MergeResponse, MultiRegionPlan, Plan, Process,
-        ProcessResponse, ResolveLockPlan, RetryRegionPlan,
+        Collect, CollectError, DefaultProcessor, Dispatch, Merge, MergeResponse, MultiRegion, Plan,
+        Process, ProcessResponse, ResolveLock, RetryRegion,
     },
     plan_builder::{PlanBuilder, SingleKey},
-    shard::{shardable_keys, shardable_range, Shardable},
+    shard::Shardable,
 };
 
 mod plan;
 mod plan_builder;
+#[macro_use]
 mod shard;
 
 /// Abstracts any request sent to a TiKV server.