Repository: seanmonstar/warp (language: rust)

Issue:
Difference between and_then and map
I'm having a little difficulty understanding the difference between the `map` and `and_then` filters.

For `Result` the difference seems to be that `and_then` replaces the entire `Result` with the function return value (if the original value is Ok) whereas `map` only replaces the Ok value. In other words you `map(foo)` is the same as `and_then(Ok(foo))`.

But in Warp they seem to be entirely different things because `and_then` is async and `map` isn't. Can you explain this? And is there a simple sync version of `and_then` that I can use instead of writing an async block without any async code?


Discussion / hints:
Hi, yeah the difference in practice is their type signature, with `and_then` you can reject, also you don't need an async block, you can use [`futures::future::ok`](https://docs.rs/futures/0.3.5/futures/future/fn.ok.html) and [`futures::future::err`](https://docs.rs/futures/0.3.5/futures/future/fn.err.html)
Thanks for the reply! I still don't quite understand `and_then` is designed to take a future and `map` isn't? I mean, why don't I need to use `futures::future::ok()` in `map()`?
yes, `and_then` has a return type of `F::Output: TryFuture + Send` so you can return an async block with async call's inside, as you stated on your comment, whereas `map` doesn't, therefore not needing nor being able to return `futures::future::ok` in a map.
Sure, but can you explain why it was *designed* like that? It seems like it is mixing separate concerns (`Result<T>/T` return type, and sync/async). Shouldn't they both return futures? Or possibly have async and non-async versions like this:

* `and_then` - returns `Result<T>`
* `and_then_async` - returns `Future<Result<T>>`
* `map` - returns `T`
* `map_async` - returns `Future<T>`
they were designed to mimic [`futures::future::FutureExt::map`](https://docs.rs/futures/0.3.5/futures/future/trait.FutureExt.html#method.map) and [`futures::future::TryFutureExt`](https://docs.rs/futures/0.3.5/futures/future/trait.TryFutureExt.html#method.and_then) applied in warp context
Aha I see, it's `futures`'s fault!

So I thought it should be like this:

| Closure returns     | `Result<T>` | `Future<Result<T>>` |
|-|-|-|
| `T`                              |  `map` | `map` |
| `Result<T>`               | `and_then` | `and_then` |
| `Future<T>`               |  -        | `map_async` |
| `Future<Result<T>>` |  -       | `and_then_async` |

But it's actually like this (maybe, this stuff is really hard to follow):

| Closure returns     | `Result<T>` | `Future<Result<T>>` |
|-|-|-|
| `T`                              |  `map` | [`map_ok`][map_ok] |
| `Result<T>`               | `and_then` | Does not exist (a) |
| `Future<T>`               |  -        | Does not exist |
| `Future<Result<T>>` |  -       | [`and_then`][and_then] |

Even more confusing there is [another `map`][map] for `Future<Result<T>>` but unlike all the other functions here its function is called if the `Result<>` is `Ok` *or* `Err`, not just for `Ok`. I think you would implement (a) by something like `x.map(|x| x.map(...))`, where the outer `map` maps the entire `Result<T>` and then the inner one maps the `Ok` value.

Would you consider changing it to be consistent for V0.3?

[map]: https://docs.rs/futures/0.3.5/futures/future/trait.FutureExt.html#method.map
[map_ok]: https://docs.rs/futures/0.3.5/futures/future/trait.TryFutureExt.html#method.map_ok
[then]: https://docs.rs/futures/0.3.5/futures/future/trait.FutureExt.html#method.then
[and_then]: https://docs.rs/futures/0.3.5/futures/future/trait.TryFutureExt.html#method.and_then

I think I just hit a snag that's related to this. I needed to use `and_then` so I could call `.await` from within the closure, but its logic was infallible, so:
```rust
    .and_then(|data| async move {
        let response = data.get().await;
        Ok(warp::reply::json(&response))
    });
```
... failed, because `warp` couldn't infer the error type as `Rejection` (and rustc threw a semi-inscrutable error). This would work:
```rust
    if true { 
        Ok(warp::reply::json(&response)) 
    } else {
        Err(warp::reject::reject())
    }
```
But that's just not right. I was finally able to divine a better solution:
```rust
    Ok::<warp::reply::Json, warp::Rejection>(warp::reply::json(&client_list))
```

An infallible async method, as @Timmmm proposes, might be more ergonomic (though less discoverable?).
Just so you know. You can use type inference to reduce some of the clutter on the return type. 

```
    Ok::<warp::reply::Json, warp::Rejection>(warp::reply::json(&client_list))
```

as

```
    Ok::<_, warp::Rejection>(warp::reply::json(&client_list))
```

I usually also have `use warp::Rejection;` at the top. So it becomes 

```
    Ok::<_, Rejection>(warp::reply::json(&client_list))
```

Not a huge difference, but I find the lack of clutter on the return types to be helpful. 
@abreis an other alternative is to move the callback to its own toplevel `async` function, which can be fully typed.

100% agree on the usefulness a `map_async` would have though, especially since a Warp rejection isn't a traditional endpoint failure, so an endpoint returning a failure would be a success from a Warp perspective.

Fix patch:
diff --git a/src/filter/and_then.rs b/src/filter/and_then.rs
index ccf78194c..22ed49c73 100644
--- a/src/filter/and_then.rs
+++ b/src/filter/and_then.rs
@@ -34,7 +34,7 @@ where
 
 #[allow(missing_debug_implementations)]
 #[pin_project]
-pub struct AndThenFuture<T: Filter, F>
+pub struct AndThenFuture<T, F>
 where
     T: Filter,
     F: Func<T::Extract>,
@@ -96,12 +96,12 @@ where
                     self.set(State::Second(fut2));
                 }
                 StateProj::Second(second) => {
-                    let ex3 = match ready!(second.try_poll(cx)) {
+                    let ex2 = match ready!(second.try_poll(cx)) {
                         Ok(item) => Ok((item,)),
                         Err(err) => Err(From::from(err)),
                     };
                     self.set(State::Done);
-                    return Poll::Ready(ex3);
+                    return Poll::Ready(ex2);
                 }
                 StateProj::Done => panic!("polled after complete"),
             }
diff --git a/src/filter/mod.rs b/src/filter/mod.rs
index acf973058..5ab39080e 100644
--- a/src/filter/mod.rs
+++ b/src/filter/mod.rs
@@ -7,6 +7,7 @@ mod or;
 mod or_else;
 mod recover;
 pub(crate) mod service;
+mod then;
 mod unify;
 mod untuple_one;
 mod wrap;
@@ -27,6 +28,7 @@ pub(crate) use self::map_err::MapErr;
 pub(crate) use self::or::Or;
 use self::or_else::OrElse;
 use self::recover::Recover;
+use self::then::Then;
 use self::unify::Unify;
 use self::untuple_one::UntupleOne;
 pub use self::wrap::wrap_fn;
@@ -197,13 +199,45 @@ pub trait Filter: FilterBase {
         }
     }
 
-    /// Composes this `Filter` with a function receiving the extracted value.
+    /// Composes this `Filter` with an async function receiving
+    /// the extracted value.
+    ///
+    /// The function should return some `Future` type.
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// use warp::Filter;
+    ///
+    /// // Map `/:id`
+    /// warp::path::param().then(|id: u64| async move {
+    ///   format!("Hello #{}", id)
+    /// });
+    /// ```
+    fn then<F>(self, fun: F) -> Then<Self, F>
+    where
+        Self: Sized,
+        F: Func<Self::Extract> + Clone,
+        F::Output: Future + Send,
+    {
+        Then {
+            filter: self,
+            callback: fun,
+        }
+    }
+
+    /// Composes this `Filter` with a fallible async function receiving
+    /// the extracted value.
     ///
     /// The function should return some `TryFuture` type.
     ///
     /// The `Error` type of the return `Future` needs be a `Rejection`, which
     /// means most futures will need to have their error mapped into one.
     ///
+    /// Rejections are meant to say "this filter didn't accept the request,
+    /// maybe another can". So for application-level errors, consider using
+    /// [`Filter::then`] instead.
+    ///
     /// # Example
     ///
     /// ```
diff --git a/src/filter/then.rs b/src/filter/then.rs
new file mode 100644
index 000000000..92d821cfa
--- /dev/null
+++ b/src/filter/then.rs
@@ -0,0 +1,95 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use futures::{ready, TryFuture};
+use pin_project::pin_project;
+
+use super::{Filter, FilterBase, Func, Internal};
+
+#[derive(Clone, Copy, Debug)]
+pub struct Then<T, F> {
+    pub(super) filter: T,
+    pub(super) callback: F,
+}
+
+impl<T, F> FilterBase for Then<T, F>
+where
+    T: Filter,
+    F: Func<T::Extract> + Clone + Send,
+    F::Output: Future + Send,
+{
+    type Extract = (<F::Output as Future>::Output,);
+    type Error = T::Error;
+    type Future = ThenFuture<T, F>;
+    #[inline]
+    fn filter(&self, _: Internal) -> Self::Future {
+        ThenFuture {
+            state: State::First(self.filter.filter(Internal), self.callback.clone()),
+        }
+    }
+}
+
+#[allow(missing_debug_implementations)]
+#[pin_project]
+pub struct ThenFuture<T, F>
+where
+    T: Filter,
+    F: Func<T::Extract>,
+    F::Output: Future + Send,
+{
+    #[pin]
+    state: State<T::Future, F>,
+}
+
+#[pin_project(project = StateProj)]
+enum State<T, F>
+where
+    T: TryFuture,
+    F: Func<T::Ok>,
+    F::Output: Future + Send,
+{
+    First(#[pin] T, F),
+    Second(#[pin] F::Output),
+    Done,
+}
+
+impl<T, F> Future for ThenFuture<T, F>
+where
+    T: Filter,
+    F: Func<T::Extract>,
+    F::Output: Future + Send,
+{
+    type Output = Result<(<F::Output as Future>::Output,), T::Error>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+        self.project().state.poll(cx)
+    }
+}
+
+impl<T, F> Future for State<T, F>
+where
+    T: TryFuture,
+    F: Func<T::Ok>,
+    F::Output: Future + Send,
+{
+    type Output = Result<(<F::Output as Future>::Output,), T::Error>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+        loop {
+            match self.as_mut().project() {
+                StateProj::First(first, second) => {
+                    let ex1 = ready!(first.try_poll(cx))?;
+                    let fut2 = second.call(ex1);
+                    self.set(State::Second(fut2));
+                }
+                StateProj::Second(second) => {
+                    let ex2 = (ready!(second.poll(cx)),);
+                    self.set(State::Done);
+                    return Poll::Ready(Ok(ex2));
+                }
+                StateProj::Done => panic!("polled after complete"),
+            }
+        }
+    }
+}


Test patch: