Merge branch 'master' into use-reg-trim-rpc-scheme

Author: hello2dj

.travis.yml

@@ -6,7 +6,8 @@ os:
   # - windows # TODO: https://github.com/pingcap/kvproto/issues/355
   - osx
 rust:
-  - stable
+  # Requires nightly for now, stable can be re-enabled when 1.36 is stable.
+  # - stable
   - nightly
 env:
   global:
@@ -27,14 +28,17 @@ addons:
       - go
 
 install:
-  - if [[ $TRAVIS_RUST_VERSION == "stable" && $TRAVIS_OS_NAME == "linux" ]]; then rustup component add rustfmt; fi
-  - if [[ $TRAVIS_RUST_VERSION == "stable" && $TRAVIS_OS_NAME == "linux" ]]; then rustup component add clippy; fi
+  # FIXME(#55) Remove when Clippy is fixed in current nightly
+  - if [[ $TRAVIS_OS_NAME == "linux" ]]; then rustup toolchain install nightly-2019-05-22; fi
+  - if [[ $TRAVIS_OS_NAME == "linux" ]]; then rustup default nightly-2019-05-22; fi
+  - if [[ $TRAVIS_OS_NAME == "linux" ]]; then rustup component add rustfmt; fi
+  - if [[ $TRAVIS_OS_NAME == "linux" ]]; then rustup component add clippy; fi
   - if [[ $TRAVIS_OS_NAME == "windows" ]]; then choco install golang cmake strawberryperl protoc; fi
   - if [[ $TRAVIS_OS_NAME == "windows" ]]; then export PATH="$PATH:/c/Go/bin/:/c/Program Files/CMake/bin"; fi
 
 script:
-  - if [[ $TRAVIS_RUST_VERSION == "stable" && $TRAVIS_OS_NAME == "linux" ]]; then cargo fmt -- --check; fi
-  - if [[ $TRAVIS_RUST_VERSION == "stable" && $TRAVIS_OS_NAME == "linux" ]]; then cargo clippy -- -D clippy::all; fi
+  - if [[ $TRAVIS_OS_NAME == "linux" ]]; then cargo fmt -- --check; fi
+  - if [[ $TRAVIS_OS_NAME == "linux" ]]; then cargo clippy -- -D clippy::all; fi
   - cargo test --all -- --nocapture
   # For now we only run full integration tests on Linux. Here's why:
   # * Docker on OS X is not supported by Travis.

Cargo.toml

@@ -20,8 +20,7 @@ name = "tikv_client"
 [dependencies]
 regex = "1"
 failure = "0.1"
-futures = "0.1"
-fxhash = "0.2"
+futures-preview = { version = "0.3.0-alpha.15", features = ["compat"] }
 grpcio = { version = "0.5.0-alpha", features = [ "secure", "prost-codec" ], default-features = false }
 lazy_static = "0.2.1"
 log = "0.3.9"
@@ -41,3 +40,5 @@ features = ["push", "process"]
 [dev-dependencies]
 clap = "2.32"
 tempdir = "0.3"
+runtime = "0.3.0-alpha.3"
+runtime-tokio = "0.3.0-alpha.3"

README.md

@@ -12,18 +12,24 @@ With this crate you can easily connect to any TiKV deployment, interact with it,
 
 This is an open source (Apache 2) project hosted by the Cloud Native Computing Foundation (CNCF) and maintained by the TiKV Authors. *We'd love it if you joined us in improving this project.*
 
-## Install
+## Using the client
 
-There are no special requirements to use this. It is a Rust 2018 edition crate supporting stable and nightly.
+The TiKV client is a Rust library (crate). It requires version 1.36 of the compiler and standard libraries (which will be stable from the 4th July 2019, see below for ensuring compatibility).
 
-To use this crate in your project, add it as a dependency in the `Cargo.toml` of your Rust project:
+To use this crate in your project, add it as a dependency in your `Cargo.toml`:
 
 ```toml
 [dependencies]
 # ...Your other dependencies...
-tikv-client = "~0.1"
+tikv-client = { git = "https://github.com/tikv/client-rust.git" }
 ```
 
+The client requires a Git dependency until we can [publish it](https://github.com/tikv/client-rust/issues/32).
+
+There are [examples](examples) which show how to use the client in a Rust program.
+
+The examples and documentation use async/await syntax. This is a new feature in Rust and is currently unstable. To use async/await you'll need to add the feature flag `#![async_await]` to your crate and use a nightly compiler (see below).
+
 ## Access the documentation
 
 We recommend using the cargo-generated documentation to browse and understand the API. We've done
@@ -37,3 +43,31 @@ You can access the documentation on your machine by running the following in any
 cargo doc --package tikv-client --open
 # If it didn't work, browse file URL it tried to open with your browser.
 ```
+
+## Toolchain versions
+
+To check what version of Rust you are using, run
+
+```bash
+rustc --version
+```
+
+You'll see something like `rustc 1.36.0-nightly (a784a8022 2019-05-09)` where the `1.36.0` is the toolchain version, and `nightly` is the channel (stable/beta/nightly). To install another toolchain use
+
+```bash
+rustup toolchain install nightly
+```
+
+Where `nightly` here is the channel to add. To update your toolchains, run
+
+```bash
+rustup update
+```
+
+To build your project using a specified toolchain, use something like
+
+```bash
+cargo +nightly build
+```
+
+Where `nightly` names the toolchain (by specifying the channel, in this case).

examples/raw.rs

@@ -1,15 +1,17 @@
 // Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
 
+#![feature(async_await, await_macro)]
+
 mod common;
 
 use crate::common::parse_args;
-use futures::future::Future;
 use tikv_client::{raw::Client, Config, Key, KvPair, Result, Value};
 
 const KEY: &str = "TiKV";
 const VALUE: &str = "Rust";
 
-fn main() -> Result<()> {
+#[runtime::main(runtime_tokio::Tokio)]
+async fn main() -> Result<()> {
     // You can try running this example by passing your pd endpoints
     // (and SSL options if necessary) through command line arguments.
     let args = parse_args("raw");
@@ -25,15 +27,14 @@ fn main() -> Result<()> {
     // When we first create a client we receive a `Connect` structure which must be resolved before
     // the client is actually connected and usable.
     let unconnnected_client = Client::new(config);
-    let client = unconnnected_client.wait()?;
+    let client = unconnnected_client.await?;
 
     // Requests are created from the connected client. These calls return structures which
     // implement `Future`. This means the `Future` must be resolved before the action ever takes
     // place.
     //
     // Here we set the key `TiKV` to have the value `Rust` associated with it.
-    let put_request = client.put(KEY, VALUE);
-    put_request.wait()?; // Returns a `tikv_client::Error` on failure.
+    client.put(KEY, VALUE).await.unwrap(); // Returns a `tikv_client::Error` on failure.
     println!("Put key {:?}, value {:?}.", KEY, VALUE);
 
     // Unlike a standard Rust HashMap all calls take owned values. This is because under the hood
@@ -45,19 +46,19 @@ fn main() -> Result<()> {
     //
     // It is best to pass a `Vec<u8>` in terms of explictness and speed. `String`s and a few other
     // types are supported as well, but it all ends up as `Vec<u8>` in the end.
-    let value: Option<Value> = client.get(KEY).wait()?;
+    let value: Option<Value> = client.get(KEY).await?;
     assert_eq!(value, Some(Value::from(VALUE)));
     println!("Get key {:?} returned value {:?}.", Key::from(KEY), value);
 
     // You can also set the `ColumnFamily` used by the request.
     // This is *advanced usage* and should have some special considerations.
-    client.delete(KEY).wait().expect("Could not delete value");
+    client.delete(KEY).await.expect("Could not delete value");
     println!("Key: {:?} deleted", Key::from(KEY));
 
     // Here we check if the key has been deleted from the key-value store.
     let value: Option<Value> = client
         .get(KEY)
-        .wait()
+        .await
         .expect("Could not get just deleted entry");
     assert!(value.is_none());
 
@@ -68,13 +69,13 @@ fn main() -> Result<()> {
         KvPair::from(("k2", "v2")),
         KvPair::from(("k3", "v3")),
     ];
-    client.batch_put(pairs).wait().expect("Could not put pairs");
+    client.batch_put(pairs).await.expect("Could not put pairs");
 
     // Same thing when you want to retrieve multiple values.
     let keys = vec![Key::from("k1"), Key::from("k2")];
     let values = client
         .batch_get(keys.clone())
-        .wait()
+        .await
         .expect("Could not get values");
     println!("Found values: {:?} for keys: {:?}", values, keys);
 
@@ -85,7 +86,7 @@ fn main() -> Result<()> {
     let pairs = client
         .scan(start..=end, 10)
         .key_only()
-        .wait()
+        .await
         .expect("Could not scan");
 
     let keys: Vec<_> = pairs.into_iter().map(|p| p.key().clone()).collect();

examples/transaction.rs

@@ -1,66 +1,76 @@
 // Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
 
+#![feature(async_await, await_macro)]
+
 mod common;
 
 use crate::common::parse_args;
-use futures::{future, Future, Stream};
+use futures::{
+    future,
+    prelude::{StreamExt, TryStreamExt},
+    stream, TryFutureExt,
+};
 use std::ops::RangeBounds;
 use tikv_client::{
     transaction::{Client, IsolationLevel},
     Config, Key, KvPair, Value,
 };
 
-fn puts(client: &Client, pairs: impl IntoIterator<Item = impl Into<KvPair>>) {
+async fn puts(client: &Client, pairs: impl IntoIterator<Item = impl Into<KvPair>>) {
     let mut txn = client.begin();
-    let _: Vec<()> = future::join_all(
+    future::join_all(
         pairs
             .into_iter()
             .map(Into::into)
             .map(|p| txn.set(p.key().clone(), p.value().clone())),
     )
-    .wait()
+    .await
+    .into_iter()
+    .collect::<Result<Vec<()>, _>>()
     .expect("Could not set key value pairs");
-    txn.commit().wait().expect("Could not commit transaction");
+    txn.commit().await.expect("Could not commit transaction");
 }
 
-fn get(client: &Client, key: Key) -> Value {
+async fn get(client: &Client, key: Key) -> Value {
     let txn = client.begin();
-    txn.get(key).wait().expect("Could not get value")
+    txn.get(key).await.expect("Could not get value")
 }
 
-fn scan(client: &Client, range: impl RangeBounds<Key>, mut limit: usize) {
+// Ignore a spurious warning from rustc (https://github.com/rust-lang/rust/issues/60566).
+#[allow(unused_mut)]
+async fn scan(client: &Client, range: impl RangeBounds<Key>, mut limit: usize) {
     client
         .begin()
         .scan(range)
-        .take_while(move |_| {
-            Ok(if limit == 0 {
+        .into_stream()
+        .take_while(move |r| {
+            assert!(r.is_ok(), "Could not scan keys");
+            future::ready(if limit == 0 {
                 false
             } else {
                 limit -= 1;
                 true
             })
         })
-        .for_each(|pair| {
-            println!("{:?}", pair);
-            Ok(())
-        })
-        .wait()
-        .expect("Could not scan keys");
+        .for_each(|pair| future::ready(println!("{:?}", pair)))
+        .await;
 }
 
-fn dels(client: &Client, keys: impl IntoIterator<Item = Key>) {
+async fn dels(client: &Client, keys: impl IntoIterator<Item = Key>) {
     let mut txn = client.begin();
     txn.set_isolation_level(IsolationLevel::ReadCommitted);
-    let _: Vec<()> = keys
-        .into_iter()
-        .map(|p| {
-            txn.delete(p).wait().expect("Could not delete key");
+    let _: Vec<()> = stream::iter(keys.into_iter())
+        .then(|p| {
+            txn.delete(p)
+                .unwrap_or_else(|e| panic!("error in delete: {:?}", e))
         })
-        .collect();
-    txn.commit().wait().expect("Could not commit transaction");
+        .collect()
+        .await;
+    txn.commit().await.expect("Could not commit transaction");
 }
 
-fn main() {
+#[runtime::main(runtime_tokio::Tokio)]
+async fn main() {
     // You can try running this example by passing your pd endpoints
     // (and SSL options if necessary) through command line arguments.
     let args = parse_args("txn");
@@ -74,27 +84,27 @@ fn main() {
     };
 
     let txn = Client::new(config)
-        .wait()
+        .await
         .expect("Could not connect to tikv");
 
     // set
     let key1: Key = b"key1".to_vec().into();
     let value1: Value = b"value1".to_vec().into();
     let key2: Key = b"key2".to_vec().into();
     let value2: Value = b"value2".to_vec().into();
-    puts(&txn, vec![(key1, value1), (key2, value2)]);
+    puts(&txn, vec![(key1, value1), (key2, value2)]).await;
 
     // get
     let key1: Key = b"key1".to_vec().into();
-    let value1 = get(&txn, key1.clone());
+    let value1 = get(&txn, key1.clone()).await;
     println!("{:?}", (key1, value1));
 
     // scan
     let key1: Key = b"key1".to_vec().into();
-    scan(&txn, key1.., 10);
+    scan(&txn, key1.., 10).await;
 
     // delete
     let key1: Key = b"key1".to_vec().into();
     let key2: Key = b"key2".to_vec().into();
-    dels(&txn, vec![key1, key2]);
+    dels(&txn, vec![key1, key2]).await;
 }