Refactor RequestState etc to not need existential types

Signed-off-by: Nick Cameron <nrc@ncameron.org>

Author: nrc

examples/raw.rs

@@ -62,30 +62,32 @@ async fn main() -> Result<()> {
         .expect("Could not get just deleted entry");
     assert!(value.is_none());
 
+    // FIXME: batch commands seem to be broken due to over-large types. I think
+    // LoopFn is to blame.
     // You can ask to write multiple key-values at the same time, it is much more
     // performant because it is passed in one request to the key-value store.
-    let pairs = vec![
-        KvPair::from(("k1", "v1")),
-        KvPair::from(("k2", "v2")),
-        KvPair::from(("k3", "v3")),
-    ];
-    client.batch_put(pairs).await.expect("Could not put pairs");
+    // let pairs = vec![
+    //     KvPair::from(("k1", "v1")),
+    //     KvPair::from(("k2", "v2")),
+    //     KvPair::from(("k3", "v3")),
+    // ];
+    // 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())
-        .await
-        .expect("Could not get values");
-    println!("Found values: {:?} for keys: {:?}", values, keys);
+    // let keys = vec![Key::from("k1"), Key::from("k2")];
+    // let values = client
+    //     .batch_get(keys.clone())
+    //     .await
+    //     .expect("Could not get values");
+    // println!("Found values: {:?} for keys: {:?}", values, keys);
 
     // Scanning a range of keys is also possible giving it two bounds
     // it will returns all entries between these two.
     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

@@ -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/lib.rs

@@ -2,7 +2,6 @@
 
 // Long and nested future chains can quickly result in large generic types.
 #![type_length_limit = "16777216"]
-#![feature(existential_type)]
 #![allow(clippy::redundant_closure)]
 
 //! This crate provides a clean, ready to use client for [TiKV](https://github.com/tikv/tikv), a