Merge pull request #73 from nrc/box-futures

Refactor everything

Author: nrc

examples/raw.rs

@@ -1,6 +1,7 @@
 // Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
 
 #![feature(async_await, await_macro)]
+#![type_length_limit = "3081103"]
 
 mod common;
 
@@ -84,8 +85,8 @@ async fn main() -> Result<()> {
     let start = "k1";
     let end = "k2";
     let pairs = client
+        .with_key_only(true)
         .scan(start..=end, 10)
-        .key_only()
         .await
         .expect("Could not scan");
 

src/compat.rs

@@ -5,7 +5,7 @@
 
 use futures::prelude::*;
 use futures::task::{Context, Poll};
-use futures::try_ready;
+use futures::{ready, try_ready};
 use std::pin::Pin;
 
 /// The status of a `loop_fn` loop.
@@ -62,6 +62,83 @@ where
     }
 }
 
+pub(crate) fn stream_fn<S, T, A, F, E>(initial_state: S, mut func: F) -> LoopFn<A, F>
+where
+    F: FnMut(S) -> A,
+    A: Future<Output = Result<Option<(S, T)>, E>>,
+{
+    LoopFn {
+        future: func(initial_state),
+        func,
+    }
+}
+
+impl<S, T, A, F, E> Stream for LoopFn<A, F>
+where
+    F: FnMut(S) -> A,
+    A: Future<Output = Result<Option<(S, T)>, E>>,
+{
+    type Item = Result<T, E>;
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        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))),
+                Ok(None) => Poll::Ready(None),
+                Ok(Some((s, t))) => {
+                    this.future = (this.func)(s);
+                    Poll::Ready(Some(Ok(t)))
+                }
+            }
+        }
+    }
+}
+
+/// A future created by the `ok_and_then` method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless polled"]
+pub(crate) struct OkAndThen<A, F> {
+    future: A,
+    func: F,
+}
+
+impl<U, T, A, F, E> Future for OkAndThen<A, F>
+where
+    F: FnMut(U) -> Result<T, E>,
+    A: Future<Output = Result<U, E>>,
+{
+    type Output = Result<T, E>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<T, E>> {
+        unsafe {
+            let this = Pin::get_unchecked_mut(self);
+            let result = try_ready!(Pin::new_unchecked(&mut this.future).poll(cx));
+            Poll::Ready((this.func)(result))
+        }
+    }
+}
+
+/// An extension crate to make using our combinator functions more ergonomic.
+pub(crate) trait ClientFutureExt {
+    /// This function is similar to `map_ok` combinator. Provide a function which
+    /// is applied after the `self` future is resolved, only if that future
+    /// resolves to `Ok`. Similar to `Result::and_then`, the supplied function
+    /// must return a Result (c.f., `map_ok`, which returns the underlying type,
+    /// `T`).
+    ///
+    /// Note that unlike `and_then`, the supplied function returns a resolved
+    /// value, not a closure.
+    fn ok_and_then<U, T, F, E>(self, func: F) -> OkAndThen<Self, F>
+    where
+        F: FnMut(U) -> Result<T, E>,
+        Self: Future<Output = Result<U, E>> + Sized,
+    {
+        OkAndThen { future: self, func }
+    }
+}
+
+impl<T: TryFuture> ClientFutureExt for T {}
+
 /// Emulate `send_all`/`SendAll` from futures 0.1 since the 0.3 versions don't
 /// work with Tokio `Handle`s due to ownership differences.
 pub(crate) trait SinkCompat<I, E> {

src/kv.rs

@@ -1,5 +1,6 @@
 // Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
 
+use derive_new::new;
 use std::cmp::{Eq, PartialEq};
 use std::convert::TryFrom;
 use std::ops::{Bound, Deref, DerefMut, Range, RangeFrom, RangeInclusive};
@@ -55,20 +56,10 @@ impl<'a> fmt::Display for HexRepr<'a> {
 /// **But, you should not need to worry about all this:** Many functions which accept a `Key`
 /// accept an `Into<Key>`, which means all of the above types can be passed directly to those
 /// functions.
-#[derive(Default, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
+#[derive(new, Default, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
 pub struct Key(Vec<u8>);
 
 impl Key {
-    #[inline]
-    pub fn new(value: Vec<u8>) -> Self {
-        Key(value)
-    }
-
-    #[inline]
-    pub(crate) fn into_inner(self) -> Vec<u8> {
-        self.0
-    }
-
     #[inline]
     fn zero_terminated(&self) -> bool {
         self.0.last().map(|i| *i == 0).unwrap_or(false)
@@ -142,6 +133,12 @@ impl AsMut<[u8]> for Key {
     }
 }
 
+impl Into<Vec<u8>> for Key {
+    fn into(self) -> Vec<u8> {
+        self.0
+    }
+}
+
 impl Deref for Key {
     type Target = [u8];
 
@@ -199,21 +196,9 @@ impl fmt::Debug for Key {
 /// **But, you should not need to worry about all this:** Many functions which accept a `Value`
 /// accept an `Into<Value>`, which means all of the above types can be passed directly to those
 /// functions.
-#[derive(Default, Clone, Eq, PartialEq, Hash)]
+#[derive(new, Default, Clone, Eq, PartialEq, Hash)]
 pub struct Value(Vec<u8>);
 
-impl Value {
-    #[inline]
-    pub fn new(value: Vec<u8>) -> Self {
-        Value(value)
-    }
-
-    #[inline]
-    pub(crate) fn into_inner(self) -> Vec<u8> {
-        self.0
-    }
-}
-
 impl From<Vec<u8>> for Value {
     fn from(v: Vec<u8>) -> Self {
         Value(v)
@@ -232,6 +217,12 @@ impl From<&'static str> for Value {
     }
 }
 
+impl Into<Vec<u8>> for Value {
+    fn into(self) -> Vec<u8> {
+        self.0
+    }
+}
+
 impl Deref for Value {
     type Target = [u8];
 
@@ -290,11 +281,6 @@ impl KvPair {
         &self.1
     }
 
-    #[inline]
-    pub fn into_inner(self) -> (Key, Value) {
-        (self.0, self.1)
-    }
-
     #[inline]
     pub fn into_key(self) -> Key {
         self.0
@@ -340,6 +326,12 @@ where
     }
 }
 
+impl Into<(Key, Value)> for KvPair {
+    fn into(self) -> (Key, Value) {
+        (self.0, self.1)
+    }
+}
+
 impl fmt::Debug for KvPair {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         let KvPair(key, value) = self;

src/lib.rs

@@ -82,6 +82,13 @@ pub mod raw;
 mod rpc;
 pub mod transaction;
 
+#[macro_use]
+extern crate lazy_static;
+#[macro_use]
+extern crate log;
+#[macro_use]
+extern crate prometheus;
+
 #[doc(inline)]
 pub use crate::config::Config;
 #[doc(inline)]