Public API Request for Comments

Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "tikv-client"
+version = "0.0.0"
+keywords = ["TiKV", "KV", "distributed-systems"]
+license = "Apache-2.0"
+authors = ["The TiKV Project Developers"]
+repository = "https://github.com/tikv/client-rust"
+description = "The rust language implementation of TiKV client."
+
+[lib]
+name = "tikv_client"
+
+[dependencies]
+futures = "0.1"
+serde = "1.0"
+serde_derive = "1.0"
+quick-error = "1.2"
+grpcio = { version = "0.4", features = [ "secure" ] }

examples/raw.rs

@@ -0,0 +1,69 @@
+extern crate futures;
+extern crate tikv_client;
+
+use std::path::PathBuf;
+
+use futures::future::Future;
+use tikv_client::*;
+
+fn main() {
+    let config = Config::new(vec!["127.0.0.1:3379"]).with_security(
+        PathBuf::from("/path/to/ca.pem"),
+        PathBuf::from("/path/to/client.pem"),
+        PathBuf::from("/path/to/client-key.pem"),
+    );
+    let raw = raw::Client::new(&config)
+        .wait()
+        .expect("Could not connect to tikv");
+
+    let key: Key = b"Company".to_vec().into();
+    let value: Value = b"PingCAP".to_vec().into();
+
+    raw.put(key.clone(), value.clone())
+        .cf("test_cf")
+        .wait()
+        .expect("Could not put kv pair to tikv");
+    println!("Successfully put {:?}:{:?} to tikv", key, value);
+
+    let value = raw
+        .get(&key)
+        .cf("test_cf")
+        .wait()
+        .expect("Could not get value");
+    println!("Found val: {:?} for key: {:?}", value, key);
+
+    raw.delete(&key)
+        .cf("test_cf")
+        .wait()
+        .expect("Could not delete value");
+    println!("Key: {:?} deleted", key);
+
+    raw.get(&key)
+        .cf("test_cf")
+        .wait()
+        .expect_err("Get returned value for not existing key");
+
+    let keys = vec![b"k1".to_vec().into(), b"k2".to_vec().into()];
+
+    let values = raw
+        .batch_get(&keys)
+        .cf("test_cf")
+        .wait()
+        .expect("Could not get values");
+    println!("Found values: {:?} for keys: {:?}", values, keys);
+
+    let start: Key = b"k1".to_vec().into();
+    let end: Key = b"k2".to_vec().into();
+    raw.scan(&start..&end, 10)
+        .cf("test_cf")
+        .key_only()
+        .wait()
+        .expect("Could not scan");
+
+    let ranges = [&start..&end, &start..&end];
+    raw.batch_scan(&ranges, 10)
+        .cf("test_cf")
+        .key_only()
+        .wait()
+        .expect("Could not batch scan");
+}

examples/transaction.rs

@@ -0,0 +1,87 @@
+extern crate futures;
+extern crate tikv_client;
+
+use std::ops::RangeBounds;
+use std::path::PathBuf;
+
+use futures::{future, Future, Stream};
+use tikv_client::transaction::{Client, IsolationLevel};
+use tikv_client::*;
+
+fn puts(client: &Client, pairs: impl IntoIterator<Item = impl Into<KvPair>>) {
+    let mut txn = client.begin();
+    let _: Vec<()> = future::join_all(
+        pairs
+            .into_iter()
+            .map(Into::into)
+            .map(|p| txn.set(p.key().clone(), p.value().clone())),
+    ).wait()
+    .expect("Could not set key value pairs");
+    txn.commit().wait().expect("Could not commit transaction");
+}
+
+fn get(client: &Client, key: &Key) -> Value {
+    let txn = client.begin();
+    txn.get(key).wait().expect("Could not get value")
+}
+
+fn scan(client: &Client, range: impl RangeBounds<Key>, mut limit: usize) {
+    client
+        .begin()
+        .scan(range)
+        .take_while(move |_| {
+            Ok(if limit == 0 {
+                false
+            } else {
+                limit -= 1;
+                true
+            })
+        }).for_each(|pair| {
+            println!("{:?}", pair);
+            Ok(())
+        }).wait()
+        .expect("Could not scan keys");
+}
+
+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");
+        }).collect();
+    txn.commit().wait().expect("Could not commit transaction");
+}
+
+fn main() {
+    let config = Config::new(vec!["127.0.0.1:3379"]).with_security(
+        PathBuf::from("/path/to/ca.pem"),
+        PathBuf::from("/path/to/client.pem"),
+        PathBuf::from("/path/to/client-key.pem"),
+    );
+    let txn = Client::new(&config)
+        .wait()
+        .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)]);
+
+    // get
+    let key1: Key = b"key1".to_vec().into();
+    let value1 = get(&txn, &key1);
+    println!("{:?}", (key1, value1));
+
+    // scan
+    let key1: Key = b"key1".to_vec().into();
+    scan(&txn, key1.., 10);
+
+    // delete
+    let key1: Key = b"key1".to_vec().into();
+    let key2: Key = b"key2".to_vec().into();
+    dels(&txn, vec![key1, key2]);
+}

src/errors.rs

@@ -0,0 +1,76 @@
+// Copyright 2016 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::error;
+use std::result;
+
+quick_error!{
+    #[derive(Debug)]
+    pub enum Error {
+        Io(err: ::std::io::Error) {
+            from()
+            cause(err)
+            description(err.description())
+        }
+        Grpc(err: ::grpc::Error) {
+            from()
+            cause(err)
+            description(err.description())
+        }
+        Canceled(err: ::futures::sync::oneshot::Canceled) {
+            from()
+            cause(err)
+            description(err.description())
+        }
+        Other(err: Box<error::Error + Sync + Send>) {
+            from()
+            cause(err.as_ref())
+            description(err.description())
+            display("unknown error {:?}", err)
+        }
+        RegionForKeyNotFound(key: Vec<u8>) {
+            description("region is not found")
+            display("region is not found for key {:?}", key)
+        }
+        RegionNotFound(id: u64) {
+            description("region is not found")
+            display("region {:?} is not found", id)
+        }
+        NotLeader(region_id: u64) {
+            description("peer is not leader")
+            display("peer is not leader for region {:?}.", region_id)
+        }
+        StoreNotMatch {
+            description("store not match")
+            display("store not match")
+        }
+        KeyNotInRegion(key: Vec<u8>, region_id: u64, start_key: Vec<u8>, end_key: Vec<u8>) {
+            description("region is not found")
+            display("key {:?} is not in region {:?}: [{:?}, {:?})", key, region_id, start_key, end_key)
+        }
+        StaleEpoch {
+            description("stale epoch")
+            display("stale epoch")
+        }
+        ServerIsBusy(reason: String) {
+            description("server is busy")
+            display("server is busy: {:?}", reason)
+        }
+        RaftEntryTooLarge(region_id: u64, entry_size: u64) {
+            description("raft entry too large")
+            display("{:?} bytes raft entry of region {:?} is too large", entry_size, region_id)
+        }
+    }
+}
+
+pub type Result<T> = result::Result<T, Error>;

src/lib.rs

@@ -0,0 +1,111 @@
+extern crate futures;
+extern crate serde;
+#[macro_use]
+extern crate serde_derive;
+#[macro_use]
+extern crate quick_error;
+extern crate grpcio as grpc;
+
+pub mod errors;
+pub mod raw;
+pub mod transaction;
+
+use std::ops::Deref;
+use std::path::PathBuf;
+
+pub use errors::Error;
+pub use errors::Result;
+
+#[derive(Default, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Debug)]
+pub struct Key(Vec<u8>);
+#[derive(Default, Clone, Eq, PartialEq, Hash, Debug)]
+pub struct Value(Vec<u8>);
+#[derive(Default, Clone, Eq, PartialEq, Debug)]
+pub struct KvPair(Key, Value);
+
+impl Into<Key> for Vec<u8> {
+    fn into(self) -> Key {
+        Key(self)
+    }
+}
+
+impl AsRef<Key> for Key {
+    fn as_ref(&self) -> &Self {
+        self
+    }
+}
+
+impl Deref for Key {
+    type Target = Vec<u8>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl Into<Value> for Vec<u8> {
+    fn into(self) -> Value {
+        Value(self)
+    }
+}
+
+impl Deref for Value {
+    type Target = Vec<u8>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl KvPair {
+    pub fn new(key: Key, value: Value) -> Self {
+        KvPair(key, value)
+    }
+
+    pub fn key(&self) -> &Key {
+        &self.0
+    }
+
+    pub fn value(&self) -> &Value {
+        &self.1
+    }
+}
+
+impl Into<KvPair> for (Key, Value) {
+    fn into(self) -> KvPair {
+        KvPair(self.0, self.1)
+    }
+}
+
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
+#[serde(default)]
+#[serde(rename_all = "kebab-case")]
+pub struct Config {
+    pub pd_endpoints: Vec<String>,
+    pub ca_path: Option<PathBuf>,
+    pub cert_path: Option<PathBuf>,
+    pub key_path: Option<PathBuf>,
+}
+
+impl Config {
+    pub fn new(pd_endpoints: impl IntoIterator<Item = impl Into<String>>) -> Self {
+        Config {
+            pd_endpoints: pd_endpoints.into_iter().map(Into::into).collect(),
+            ca_path: None,
+            cert_path: None,
+            key_path: None,
+        }
+    }
+
+    pub fn with_security(
+        mut self,
+        ca_path: impl Into<PathBuf>,
+        cert_path: impl Into<PathBuf>,
+        key_path: impl Into<PathBuf>,
+    ) -> Self {
+        self.ca_path = Some(ca_path.into());
+        self.cert_path = Some(cert_path.into());
+        self.key_path = Some(key_path.into());
+        self
+    }
+}