Repository: actix/actix (language: rust)

Issue:
`TryFutureExt` functions for `ActorFuture`
The `TryFutureExt` extension trait offers functions like `and_then` and `map_err` for futures that return `Result` types. It would be great to have these functions available on `ActorFuture` as well.


Discussion / hints:
These combinators are really useful, I see `and_then` and `map_err` was removed since 0.9 release. What is the background, do we have further plan?

Isn't it the case they were removed in the same way that std Future trait just deals with a single Output associated type now instead of an Ok and an Err.

Given that, isn't it trivial to emulate `and_then` with `ActorFuture::map` + `Result::and_then` and similarly `map_err` with `ActorFuture::map` + `Result::map_err`?

This would likely require a separate trait (`ActorTryFuture`?) if we want to build these in.
@robjtede From our cases, yes it is a bit trivial. And `TryStreamExt` for `ActorStream` is the same.

Fix patch:
diff --git a/actix/CHANGES.md b/actix/CHANGES.md
index 930b64235..eefcfd644 100644
--- a/actix/CHANGES.md
+++ b/actix/CHANGES.md
@@ -1,6 +1,11 @@
 # CHANGES
 
 ## Unreleased - 2021-xx-xx
+### Added
+* Add `fut::try_future::ActorTryFuture`. [#419]
+* Add `fut::try_future::ActorTryFutureExt` trait with `map_ok`, `map_err` and `and_then` combinator. [#419] 
+
+[#419]: https://github.com/actix/actix/pull/419
 
 
 ## 0.11.1 - 2021-03-23
diff --git a/actix/src/fut/future/then.rs b/actix/src/fut/future/then.rs
index 215194dff..fa0b6972c 100644
--- a/actix/src/fut/future/then.rs
+++ b/actix/src/fut/future/then.rs
@@ -11,7 +11,7 @@ pin_project! {
     /// Future for the `then` combinator, chaining computations on the end of
     /// another future regardless of its outcome.
     ///
-    /// This is created by the `Future::then` method.
+    /// This is created by the `ActorFuture::then` method.
     #[project = ThenProj]
     #[derive(Debug)]
     #[must_use = "futures do nothing unless polled"]
diff --git a/actix/src/fut/mod.rs b/actix/src/fut/mod.rs
index 861f4e077..40c981185 100644
--- a/actix/src/fut/mod.rs
+++ b/actix/src/fut/mod.rs
@@ -2,9 +2,11 @@
 
 pub mod future;
 pub mod stream;
+pub mod try_future;
 
 pub use self::future::{
     result::{err, ok, ready, result, Ready},
     wrap_future, ActorFuture, ActorFutureExt, LocalBoxActorFuture, WrapFuture,
 };
 pub use self::stream::{wrap_stream, ActorStream, ActorStreamExt, WrapStream};
+pub use self::try_future::{ActorTryFuture, ActorTryFutureExt};
diff --git a/actix/src/fut/try_future/and_then.rs b/actix/src/fut/try_future/and_then.rs
new file mode 100644
index 000000000..fdf0b8319
--- /dev/null
+++ b/actix/src/fut/try_future/and_then.rs
@@ -0,0 +1,77 @@
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use futures_util::ready;
+use pin_project_lite::pin_project;
+
+use crate::actor::Actor;
+use crate::fut::{future::ActorFuture, try_future::ActorTryFuture};
+
+pin_project! {
+    /// Future for the `and_then` combinator, chaining computations
+    /// on the end of another actor future regardless of its outcome.
+    ///
+    /// This is created by the `ActorTryFuture::and_then` method.
+    #[project = AndThenProj]
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless polled"]
+    pub enum AndThen<A, B, F> {
+        First {
+            #[pin]
+            fut1: A,
+            data: Option<F>,
+        },
+        Second {
+            #[pin]
+            fut2: B
+        },
+        Empty,
+    }
+}
+
+pub(super) fn new<A, B, F, Act>(future: A, f: F) -> AndThen<A, B, F>
+where
+    A: ActorTryFuture<Act>,
+    B: ActorTryFuture<Act>,
+    Act: Actor,
+{
+    AndThen::First {
+        fut1: future,
+        data: Some(f),
+    }
+}
+
+impl<A, B, F, Act> ActorFuture<Act> for AndThen<A, B, F>
+where
+    A: ActorTryFuture<Act>,
+    B: ActorTryFuture<Act, Error = A::Error>,
+    F: FnOnce(A::Ok, &mut Act, &mut Act::Context) -> B,
+    Act: Actor,
+{
+    type Output = Result<B::Ok, A::Error>;
+
+    fn poll(
+        mut self: Pin<&mut Self>,
+        act: &mut Act,
+        ctx: &mut Act::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Self::Output> {
+        match self.as_mut().project() {
+            AndThenProj::First { fut1, data } => {
+                let ok = ready!(fut1.try_poll(act, ctx, task))?;
+                let data = data.take().unwrap();
+                let fut2 = data(ok, act, ctx);
+                self.set(AndThen::Second { fut2 });
+                self.poll(act, ctx, task)
+            }
+            AndThenProj::Second { fut2 } => {
+                let res = ready!(fut2.try_poll(act, ctx, task));
+                self.set(AndThen::Empty);
+                Poll::Ready(res)
+            }
+            AndThenProj::Empty => panic!("ActorFuture polled after finish"),
+        }
+    }
+}
diff --git a/actix/src/fut/try_future/map_err.rs b/actix/src/fut/try_future/map_err.rs
new file mode 100644
index 000000000..1babc9169
--- /dev/null
+++ b/actix/src/fut/try_future/map_err.rs
@@ -0,0 +1,65 @@
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use futures_core::ready;
+use pin_project_lite::pin_project;
+
+use crate::{
+    fut::{future::ActorFuture, try_future::ActorTryFuture},
+    Actor,
+};
+
+pin_project! {
+    /// Future for the [`map`](super::ActorTryFutureExt::map_err) method.
+    #[project = MapProj]
+    #[project_replace = MapProjReplace]
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub enum MapErr<Fut, F> {
+        Incomplete {
+            #[pin]
+            future: Fut,
+            f: F,
+        },
+        Complete,
+    }
+}
+
+impl<Fut, F> MapErr<Fut, F> {
+    pub(crate) fn new(future: Fut, f: F) -> Self {
+        Self::Incomplete { future, f }
+    }
+}
+
+impl<U, Fut, A, F> ActorFuture<A> for MapErr<Fut, F>
+where
+    Fut: ActorTryFuture<A>,
+    A: Actor,
+    F: FnOnce(Fut::Error, &mut A, &mut A::Context) -> U,
+{
+    type Output = Result<Fut::Ok, U>;
+
+    fn poll(
+        mut self: Pin<&mut Self>,
+        act: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Self::Output> {
+        match self.as_mut().project() {
+            MapProj::Incomplete { future, .. } => {
+                let output = ready!(future.try_poll(act, ctx, task));
+                match self.project_replace(MapErr::Complete) {
+                    MapProjReplace::Incomplete { f, .. } => {
+                        Poll::Ready(output.map_err(|err| f(err, act, ctx)))
+                    }
+                    MapProjReplace::Complete => unreachable!(),
+                }
+            }
+            MapProj::Complete => {
+                panic!("MapErr must not be polled after it returned `Poll::Ready`")
+            }
+        }
+    }
+}
diff --git a/actix/src/fut/try_future/map_ok.rs b/actix/src/fut/try_future/map_ok.rs
new file mode 100644
index 000000000..bfe7fa987
--- /dev/null
+++ b/actix/src/fut/try_future/map_ok.rs
@@ -0,0 +1,65 @@
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use futures_core::ready;
+use pin_project_lite::pin_project;
+
+use crate::{
+    fut::{future::ActorFuture, try_future::ActorTryFuture},
+    Actor,
+};
+
+pin_project! {
+    /// Future for the [`map`](super::ActorTryFutureExt::map_ok) method.
+    #[project = MapProj]
+    #[project_replace = MapProjReplace]
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub enum MapOk<Fut, F> {
+        Incomplete {
+            #[pin]
+            future: Fut,
+            f: F,
+        },
+        Complete,
+    }
+}
+
+impl<Fut, F> MapOk<Fut, F> {
+    pub(crate) fn new(future: Fut, f: F) -> Self {
+        Self::Incomplete { future, f }
+    }
+}
+
+impl<U, Fut, A, F> ActorFuture<A> for MapOk<Fut, F>
+where
+    Fut: ActorTryFuture<A>,
+    A: Actor,
+    F: FnOnce(Fut::Ok, &mut A, &mut A::Context) -> U,
+{
+    type Output = Result<U, Fut::Error>;
+
+    fn poll(
+        mut self: Pin<&mut Self>,
+        act: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Self::Output> {
+        match self.as_mut().project() {
+            MapProj::Incomplete { future, .. } => {
+                let output = ready!(future.try_poll(act, ctx, task));
+                match self.project_replace(MapOk::Complete) {
+                    MapProjReplace::Incomplete { f, .. } => {
+                        Poll::Ready(output.map(|ok| f(ok, act, ctx)))
+                    }
+                    MapProjReplace::Complete => unreachable!(),
+                }
+            }
+            MapProj::Complete => {
+                panic!("MapOk must not be polled after it returned `Poll::Ready`")
+            }
+        }
+    }
+}
diff --git a/actix/src/fut/try_future/mod.rs b/actix/src/fut/try_future/mod.rs
new file mode 100644
index 000000000..85d72895d
--- /dev/null
+++ b/actix/src/fut/try_future/mod.rs
@@ -0,0 +1,139 @@
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use crate::{actor::Actor, fut::future::ActorFuture};
+
+mod and_then;
+mod map_err;
+mod map_ok;
+
+pub use and_then::AndThen;
+pub use map_err::MapErr;
+pub use map_ok::MapOk;
+
+mod private_try_act_future {
+    use super::{Actor, ActorFuture};
+
+    pub trait Sealed<A> {}
+
+    impl<A, F, T, E> Sealed<A> for F
+    where
+        A: Actor,
+        F: ?Sized + ActorFuture<A, Output = Result<T, E>>,
+    {
+    }
+}
+
+/// A convenience for actor futures that return `Result` values that includes
+/// a variety of adapters tailored to such actor futures.
+pub trait ActorTryFuture<A: Actor>: ActorFuture<A> + private_try_act_future::Sealed<A> {
+    /// The type of successful values yielded by this actor future
+    type Ok;
+
+    /// The type of failures yielded by this actor  future
+    type Error;
+
+    /// Poll this `ActorTryFuture` as if it were a `ActorFuture`.
+    ///
+    /// This method is a stopgap for a compiler limitation that prevents us from
+    /// directly inheriting from the `ActorFuture` trait; in the actor future it
+    /// won't be needed.
+    fn try_poll(
+        self: Pin<&mut Self>,
+        srv: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Result<Self::Ok, Self::Error>>;
+}
+
+impl<A, F, T, E> ActorTryFuture<A> for F
+where
+    A: Actor,
+    F: ActorFuture<A, Output = Result<T, E>> + ?Sized,
+{
+    type Ok = T;
+    type Error = E;
+
+    fn try_poll(
+        self: Pin<&mut Self>,
+        srv: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<F::Output> {
+        self.poll(srv, ctx, task)
+    }
+}
+
+/// Adapters specific to [`Result`]-returning actor futures
+pub trait ActorTryFutureExt<A: Actor>: ActorTryFuture<A> {
+    /// Executes another actor future after this one resolves successfully.
+    /// The success value is passed to a closure to create this subsequent
+    /// actor future.
+    ///
+    /// The provided closure `f` will only be called if this actor future is
+    /// resolved to an [`Ok`]. If this actor future resolves to an  [`Err`],
+    /// panics, or is dropped, then the provided closure will never
+    /// be invoked. The [`Error`](ActorTryFuture::Error) type of
+    /// this actor future and the actor future returned by `f` have to match.
+    ///
+    /// Note that this method consumes the actor future it is called on and
+    /// returns a wrapped version of it.
+    fn and_then<Fut, F>(self, f: F) -> AndThen<Self, Fut, F>
+    where
+        F: FnOnce(Self::Ok, &mut A, &mut A::Context) -> Fut,
+        Fut: ActorTryFuture<A, Error = Self::Error>,
+        Self: Sized,
+    {
+        and_then::new(self, f)
+    }
+
+    /// Maps this actor future's success value to a different value.
+    ///
+    /// This method can be used to change the [`Ok`](ActorTryFuture::Ok) type
+    /// of the actor future into a different type. It is similar to the
+    /// [`Result::map`]  method. You can use this method to chain along a
+    /// computation once the actor future has been resolved.
+    ///
+    /// The provided closure `f` will only be called if this actor future is
+    /// resolved to an [`Ok`]. If it resolves to an [`Err`], panics, or is
+    /// dropped, then the provided closure will never be invoked.
+    ///
+    /// Note that this method consumes the actor future it is called on and
+    /// returns a wrapped version of it.
+    fn map_ok<T, F>(self, f: F) -> MapOk<Self, F>
+    where
+        F: FnOnce(Self::Ok, &mut A, &mut A::Context) -> T,
+        Self: Sized,
+    {
+        MapOk::new(self, f)
+    }
+
+    /// Maps this actor future's error value to a different value.
+    ///
+    /// This method can be used to change the [`Error`](ActorTryFuture::Error)
+    /// type of the actor future into a different type. It is similar to the
+    /// [`Result::map_err`] method.
+    ///
+    /// The provided closure `f` will only be called if this actor future is
+    /// resolved to an [`Err`]. If it resolves to an [`Ok`], panics, or is
+    /// dropped, then the provided closure will never be invoked.
+    ///
+    /// Note that this method consumes the actor future it is called on and
+    /// returns a wrapped version of it.
+    fn map_err<T, F>(self, f: F) -> MapErr<Self, F>
+    where
+        F: FnOnce(Self::Error, &mut A, &mut A::Context) -> T,
+        Self: Sized,
+    {
+        MapErr::new(self, f)
+    }
+}
+
+impl<A, F> ActorTryFutureExt<A> for F
+where
+    A: Actor,
+    F: ActorTryFuture<A> + ?Sized,
+{
+}
diff --git a/actix/src/lib.rs b/actix/src/lib.rs
index 42b37a46b..c92644bf8 100644
--- a/actix/src/lib.rs
+++ b/actix/src/lib.rs
@@ -61,7 +61,8 @@ pub use crate::actor::{
 pub use crate::address::{Addr, MailboxError, Recipient, WeakAddr, WeakRecipient};
 pub use crate::context::Context;
 pub use crate::fut::{
-    ActorFuture, ActorFutureExt, ActorStream, ActorStreamExt, WrapFuture, WrapStream,
+    ActorFuture, ActorFutureExt, ActorStream, ActorStreamExt, ActorTryFuture,
+    ActorTryFutureExt, WrapFuture, WrapStream,
 };
 pub use crate::handler::{
     ActorResponse, AtomicResponse, Handler, Message, MessageResult, Response,
@@ -100,7 +101,8 @@ pub mod prelude {
     };
     pub use crate::context::{Context, ContextFutureSpawner};
     pub use crate::fut::{
-        ActorFuture, ActorFutureExt, ActorStream, ActorStreamExt, WrapFuture, WrapStream,
+        ActorFuture, ActorFutureExt, ActorStream, ActorStreamExt, ActorTryFuture,
+        ActorTryFutureExt, WrapFuture, WrapStream,
     };
     pub use crate::handler::{
         ActorResponse, AtomicResponse, Handler, Message, MessageResult, Response,


Test patch:
diff --git a/actix/tests/test_fut.rs b/actix/tests/test_fut.rs
index 7cb112e50..8ad041dc2 100644
--- a/actix/tests/test_fut.rs
+++ b/actix/tests/test_fut.rs
@@ -177,6 +177,39 @@ impl Handler<CollectMsg> for MyStreamActor2 {
     }
 }
 
+struct TryFutureMsg(usize);
+
+impl Message for TryFutureMsg {
+    type Result = Result<isize, u32>;
+}
+
+impl Handler<TryFutureMsg> for MyStreamActor2 {
+    type Result = ResponseActFuture<Self, Result<isize, u32>>;
+
+    fn handle(&mut self, msg: TryFutureMsg, _: &mut Context<Self>) -> Self::Result {
+        let num = msg.0;
+        Box::pin(
+            async move {
+                assert_eq!(num, 5);
+                Ok::<usize, usize>(num * 2)
+            }
+            .into_actor(self)
+            .map_ok(|res, _, _| {
+                assert_eq!(10usize, res);
+                res as isize
+            })
+            .and_then(|res, _, _| {
+                assert_eq!(10isize, res);
+                fut::err::<isize, usize>(996)
+            })
+            .map_err(|e, _, _| {
+                assert_eq!(996usize, e);
+                e as u32
+            }),
+        )
+    }
+}
+
 #[test]
 fn test_stream_timeout() {
     let timeout = Arc::new(AtomicBool::new(false));
@@ -216,3 +249,12 @@ fn test_stream_collect() {
         assert_eq!(res, vec![1, 2, 3, 4, 5]);
     })
 }
+
+#[test]
+fn test_try_future() {
+    System::new().block_on(async {
+        let addr = MyStreamActor2 { counter: 0 }.start();
+        let res = addr.send(TryFutureMsg(5)).await.unwrap();
+        assert_eq!(res.err().unwrap(), 996u32);
+    })
+}