Repository: actix/actix (language: rust)

Issue:
ActorStream is missing some important Stream combinators
Hi,
There are some important `Stream` combinators, which are not supported by `ActorStream` right now. Especially `take_while` and `skip_while` would be nice to have access to `Actor` and `Context` instances in them. This would allow for better control over never ending streams like `Iterator` for example (e.g `take_while` would be able to access `Actor`'s field directly to check if the stream needs to be stopped).


Discussion / hints:

Fix patch:
diff --git a/actix/CHANGES.md b/actix/CHANGES.md
index db762e86f..2f2adeb3e 100644
--- a/actix/CHANGES.md
+++ b/actix/CHANGES.md
@@ -1,14 +1,30 @@
 # CHANGES
 
 ## Unreleased - 2021-xx-xx
+### Added
+* Added `actix::fut::{ActorFutureExt, ActorStreamExt}` traits for extension method for `ActorFuture` 
+  and `ActorStream` trait. This is aiming to have a similar traits set inline with `futures` crate. [#474]
+* Added `ActorStreamExt::collect` method for collect an actor stream's items and output them as an
+  actor future. [#474]
+* Added `ActorStreamExt::take_while` method to take an actor stream's items based on the closure output. [#474]
+* Added `ActorStreamExt::skip_while` method to skip an actor stream's items based on the closure output. [#474]
+* Added `actix::fut::LocalBoxActorFuture` type to keep inline with `futures::future::LocalBoxFuture` type. [#474]
+
 ### Changed
 * Rework `ActorFuture` trait. [#465]
 * `actix::fut::{wrap_future, wrap_stream}` would need type annotation for Actor type. [#465]
-* Remove `dev::ResponseChannel` trait [#472]
 * `dev::MessageResponse::handle` method does not need generic type [#472]
+* `actix::fut::{ok, err, result, FutureResult, Either}` are changed to re-export of 
+  `futures::future::{ready, Ready, Either}` types. [#474]
+* `futures::future::{ok, err, ready, Ready, Either}` types impls `ActorFuture` trait by default. [#474]
+  
+
+### Removed
+* Remove `dev::ResponseChannel` trait [#472]
 
 [#465]: https://github.com/actix/actix/issues/465
 [#472]: https://github.com/actix/actix/pull/472
+[#474]: https://github.com/actix/actix/pull/474
 
 
 ## 0.11.0-beta.2 - 2021-02-06
diff --git a/actix/src/actor.rs b/actix/src/actor.rs
index 1b753171e..c55160ae5 100644
--- a/actix/src/actor.rs
+++ b/actix/src/actor.rs
@@ -7,7 +7,7 @@ use log::error;
 use crate::address::{channel, Addr};
 use crate::context::Context;
 use crate::contextitems::{ActorDelayedMessageItem, ActorMessageItem, ActorMessageStreamItem};
-use crate::fut::{ActorFuture, ActorStream};
+use crate::fut::{ActorFuture, ActorStreamExt};
 use crate::handler::{Handler, Message};
 use crate::mailbox::DEFAULT_CAPACITY;
 use crate::stream::StreamHandler;
diff --git a/actix/src/actors/resolver.rs b/actix/src/actors/resolver.rs
index 035648425..a2387163e 100644
--- a/actix/src/actors/resolver.rs
+++ b/actix/src/actors/resolver.rs
@@ -227,8 +227,8 @@ impl Handler<Connect> for Resolver {
                 self.resolver.as_ref().unwrap(),
             )
             .then(move |addrs, act, _| match addrs {
-                Ok(a) => Either::left(TcpConnector::with_timeout(a, timeout)),
-                Err(e) => Either::right(async move { Err(e) }.into_actor(act)),
+                Ok(a) => Either::Left(TcpConnector::with_timeout(a, timeout)),
+                Err(e) => Either::Right(async move { Err(e) }.into_actor(act)),
             }),
         )
     }
diff --git a/actix/src/fut/chain.rs b/actix/src/fut/chain.rs
deleted file mode 100644
index 1dcfad05f..000000000
--- a/actix/src/fut/chain.rs
+++ /dev/null
@@ -1,76 +0,0 @@
-use std::marker::PhantomData;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-
-use pin_project_lite::pin_project;
-
-use crate::actor::Actor;
-use crate::fut::ActorFuture;
-
-pin_project! {
-    #[project = ChainProj]
-    #[must_use = "futures do nothing unless you `.await` or poll them"]
-    #[derive(Debug)]
-    pub enum Chain<A, B, Fn, Act> {
-        First {
-            #[pin]
-            fut1: A,
-            data: Option<Fn>,
-            _act: PhantomData<Act>,
-        },
-        Second {
-            #[pin]
-            fut2: B
-        },
-        Empty,
-    }
-}
-
-impl<A, B, Fn, Act> Chain<A, B, Fn, Act>
-where
-    A: ActorFuture<Act>,
-    B: ActorFuture<Act>,
-    Act: Actor,
-{
-    pub fn new(fut1: A, data: Fn) -> Self {
-        Chain::First {
-            fut1,
-            data: Some(data),
-            _act: PhantomData,
-        }
-    }
-
-    pub fn poll<Fn1>(
-        mut self: Pin<&mut Self>,
-        srv: &mut Act,
-        ctx: &mut Act::Context,
-        task: &mut Context<'_>,
-        f: Fn1,
-    ) -> Poll<B::Output>
-    where
-        Fn1: FnOnce(A::Output, Fn, &mut Act, &mut Act::Context) -> B,
-    {
-        let mut f = Some(f);
-
-        loop {
-            let this = self.as_mut().project();
-            let (output, data) = match this {
-                ChainProj::First { fut1, data, .. } => {
-                    let output = match fut1.poll(srv, ctx, task) {
-                        Poll::Ready(t) => t,
-                        Poll::Pending => return Poll::Pending,
-                    };
-                    (output, data.take().unwrap())
-                }
-                ChainProj::Second { fut2 } => {
-                    return fut2.poll(srv, ctx, task);
-                }
-                ChainProj::Empty => unreachable!(),
-            };
-
-            self.set(Chain::Empty);
-            let fut2 = (f.take().unwrap())(output, data, srv, ctx);
-            self.set(Chain::Second { fut2 })
-        }
-    }
-}
diff --git a/actix/src/fut/either.rs b/actix/src/fut/either.rs
deleted file mode 100644
index d00ff45a3..000000000
--- a/actix/src/fut/either.rs
+++ /dev/null
@@ -1,93 +0,0 @@
-use std::pin::Pin;
-use std::task::{Context, Poll};
-
-use pin_project_lite::pin_project;
-
-use crate::actor::Actor;
-use crate::fut::ActorFuture;
-
-pin_project! {
-    /// Combines two different futures yielding the same item and error
-    /// types into a single type.
-    #[project = EitherProj]
-    #[derive(Debug)]
-    pub enum Either<A, B> {
-        /// First branch of the type
-        Left {
-            #[pin]
-            left: A
-        },
-        /// Second branch of the type
-        Right {
-            #[pin]
-            right: B
-        },
-    }
-}
-
-impl<A, B> Either<A, B> {
-    /// construct first branch of the type
-    pub fn left(left: A) -> Self {
-        Self::Left { left }
-    }
-
-    /// construct second branch of the type
-    pub fn right(right: B) -> Self {
-        Self::Right { right }
-    }
-}
-
-impl<A, B, T> Either<(T, A), (T, B)> {
-    /// Factor out a homogeneous type from an either of pairs.
-    ///
-    /// Here, the homogeneous type is the first element of the pairs.
-    pub fn factor_left(self) -> (T, Either<A, B>) {
-        match self {
-            Either::Left { left: (x, left) } => (x, Either::Left { left }),
-            Either::Right { right: (x, right) } => (x, Either::Right { right }),
-        }
-    }
-}
-
-impl<A, B, T> Either<(A, T), (B, T)> {
-    /// Factor out a homogeneous type from an either of pairs.
-    ///
-    /// Here, the homogeneous type is the second element of the pairs.
-    pub fn factor_right(self) -> (Either<A, B>, T) {
-        match self {
-            Either::Left { left: (left, x) } => (Either::Left { left }, x),
-            Either::Right { right: (right, x) } => (Either::Right { right }, x),
-        }
-    }
-}
-
-impl<T> Either<T, T> {
-    /// Extract the value of an either over two equivalent types.
-    pub fn into_inner(self) -> T {
-        match self {
-            Either::Left { left } => left,
-            Either::Right { right } => right,
-        }
-    }
-}
-
-impl<A, Act, B> ActorFuture<Act> for Either<A, B>
-where
-    Act: Actor,
-    A: ActorFuture<Act>,
-    B: ActorFuture<Act, Output = A::Output>,
-{
-    type Output = A::Output;
-
-    fn poll(
-        self: Pin<&mut Self>,
-        act: &mut Act,
-        ctx: &mut Act::Context,
-        task: &mut Context<'_>,
-    ) -> Poll<A::Output> {
-        match self.project() {
-            EitherProj::Left { left } => left.poll(act, ctx, task),
-            EitherProj::Right { right } => right.poll(act, ctx, task),
-        }
-    }
-}
diff --git a/actix/src/fut/future/either.rs b/actix/src/fut/future/either.rs
new file mode 100644
index 000000000..991692929
--- /dev/null
+++ b/actix/src/fut/future/either.rs
@@ -0,0 +1,41 @@
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+// TODO: remove re-export and encourage direct usage of futures crate types.
+pub use futures_util::future::Either;
+
+use crate::actor::Actor;
+use crate::fut::ActorFuture;
+
+impl<A, B, Act> ActorFuture<Act> for Either<A, B>
+where
+    A: ActorFuture<Act>,
+    B: ActorFuture<Act, Output = A::Output>,
+    Act: Actor,
+{
+    type Output = A::Output;
+
+    fn poll(
+        self: Pin<&mut Self>,
+        act: &mut Act,
+        ctx: &mut Act::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<A::Output> {
+        // SAFETY:
+        //
+        // Copied from futures_util::future::Either::project method.
+        // This is used to expose this method to public.
+        // It has the same safety as the private one.
+        let this = unsafe {
+            match self.get_unchecked_mut() {
+                Either::Left(a) => Either::Left(Pin::new_unchecked(a)),
+                Either::Right(b) => Either::Right(Pin::new_unchecked(b)),
+            }
+        };
+
+        match this {
+            Either::Left(left) => left.poll(act, ctx, task),
+            Either::Right(right) => right.poll(act, ctx, task),
+        }
+    }
+}
diff --git a/actix/src/fut/future/map.rs b/actix/src/fut/future/map.rs
new file mode 100644
index 000000000..577095c87
--- /dev/null
+++ b/actix/src/fut/future/map.rs
@@ -0,0 +1,59 @@
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use futures_core::ready;
+use pin_project_lite::pin_project;
+
+use crate::actor::Actor;
+use crate::fut::ActorFuture;
+
+pin_project! {
+    /// Future for the [`map`](super::ActorFutureExt::map) method.
+    #[project = MapProj]
+    #[project_replace = MapProjReplace]
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub enum Map<Fut, Fn> {
+        Incomplete {
+            #[pin]
+            future: Fut,
+            f: Fn,
+        },
+        Complete,
+    }
+}
+
+impl<Fut, Fn> Map<Fut, Fn> {
+    pub(super) fn new(future: Fut, f: Fn) -> Self {
+        Self::Incomplete { future, f }
+    }
+}
+
+impl<U, Fut, A, Fn> ActorFuture<A> for Map<Fut, Fn>
+where
+    Fut: ActorFuture<A>,
+    A: Actor,
+    Fn: FnOnce(Fut::Output, &mut A, &mut A::Context) -> U,
+{
+    type Output = 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.poll(act, ctx, task));
+                match self.project_replace(Map::Complete) {
+                    MapProjReplace::Incomplete { f, .. } => Poll::Ready(f(output, act, ctx)),
+                    MapProjReplace::Complete => unreachable!(),
+                }
+            }
+            MapProj::Complete => {
+                panic!("Map must not be polled after it returned `Poll::Ready`")
+            }
+        }
+    }
+}
diff --git a/actix/src/fut/future/mod.rs b/actix/src/fut/future/mod.rs
new file mode 100644
index 000000000..2b259299f
--- /dev/null
+++ b/actix/src/fut/future/mod.rs
@@ -0,0 +1,312 @@
+use std::{
+    future::Future,
+    marker::PhantomData,
+    ops::{Deref, DerefMut},
+    pin::Pin,
+    task::{Context, Poll},
+    time::Duration,
+};
+
+use pin_project_lite::pin_project;
+
+pub use either::Either;
+pub use map::Map;
+pub use then::Then;
+pub use timeout::Timeout;
+
+use crate::actor::Actor;
+
+mod either;
+mod map;
+pub mod result;
+mod then;
+mod timeout;
+
+/// Trait for types which are a placeholder of a value that may become
+/// available at some later point in time.
+///
+/// `ActorFuture` is very similar to a regular `Future`, only with subsequent combinator closures accepting the actor and its context, in addition to the result.
+///
+/// `ActorFuture` allows for use cases where future processing requires access to the actor or its context.
+///
+/// Here is an example of a handler on a single actor, deferring work to another actor, and
+/// then updating the initiating actor's state:
+///
+/// ```rust,no_run
+/// use actix::prelude::*;
+///
+/// // The response type returned by the actor future
+/// type OriginalActorResponse = ();
+/// // The error type returned by the actor future
+/// type MessageError = ();
+/// // This is the needed result for the DeferredWork message
+/// // It's a result that combine both Response and Error from the future response.
+/// type DeferredWorkResult = Result<OriginalActorResponse, MessageError>;
+/// #
+/// # struct ActorState {}
+/// #
+/// # impl ActorState {
+/// #    fn update_from(&mut self, _result: ()) {}
+/// # }
+/// #
+/// # struct OtherActor {}
+/// #
+/// # impl Actor for OtherActor {
+/// #    type Context = Context<Self>;
+/// # }
+/// #
+/// # impl Handler<OtherMessage> for OtherActor {
+/// #    type Result = ();
+/// #
+/// #    fn handle(&mut self, _msg: OtherMessage, _ctx: &mut Context<Self>) -> Self::Result {
+/// #    }
+/// # }
+/// #
+/// # struct OriginalActor{
+/// #     other_actor: Addr<OtherActor>,
+/// #     inner_state: ActorState
+/// # }
+/// #
+/// # impl Actor for OriginalActor{
+/// #     type Context = Context<Self>;
+/// # }
+/// #
+/// # #[derive(Message)]
+/// # #[rtype(result = "Result<(), MessageError>")]
+/// # struct DeferredWork{}
+/// #
+/// # #[derive(Message)]
+/// # #[rtype(result = "()")]
+/// # struct OtherMessage{}
+///
+/// impl Handler<DeferredWork> for OriginalActor {
+///     // Notice the `Response` is an `ActorFuture`-ized version of `Self::Message::Result`.
+///     type Result = ResponseActFuture<Self, Result<OriginalActorResponse, MessageError>>;
+///
+///     fn handle(&mut self, _msg: DeferredWork, _ctx: &mut Context<Self>) -> Self::Result {
+///         // this creates a `Future` representing the `.send` and subsequent `Result` from
+///         // `other_actor`
+///         let send_to_other = self.other_actor
+///             .send(OtherMessage {});
+///
+///         // Wrap that `Future` so subsequent chained handlers can access
+///         // the `actor` (`self` in the  synchronous code) as well as the context.
+///         let send_to_other = actix::fut::wrap_future::<_, Self>(send_to_other);
+///
+///         // once the wrapped future resolves, update this actor's state
+///         let update_self = send_to_other.map(|result, actor, _ctx| {
+///             // Actor's state updated here
+///             match result {
+///                 Ok(v) => {
+///                     actor.inner_state.update_from(v);
+///                     Ok(())
+///                 },
+///                 // Failed to send message to other_actor
+///                 Err(_e) => Err(()),
+///             }
+///         });
+///
+///         // return the wrapped future
+///         Box::pin(update_self)
+///     }
+/// }
+///
+/// ```
+///
+/// See also [into_actor](trait.WrapFuture.html#tymethod.into_actor), which provides future conversion using trait
+pub trait ActorFuture<A: Actor> {
+    /// The type of value that this future will resolved with if it is
+    /// successful.
+    type Output;
+
+    fn poll(
+        self: Pin<&mut Self>,
+        srv: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Self::Output>;
+}
+
+pub trait ActorFutureExt<A: Actor>: ActorFuture<A> {
+    /// Map this future's result to a different type, returning a new future of
+    /// the resulting type.
+    fn map<Fn, U>(self, f: Fn) -> Map<Self, Fn>
+    where
+        Fn: FnOnce(Self::Output, &mut A, &mut A::Context) -> U,
+        Self: Sized,
+    {
+        Map::new(self, f)
+    }
+
+    /// Chain on a computation for when a future finished, passing the result of
+    /// the future to the provided closure `f`.
+    fn then<Fn, Fut>(self, f: Fn) -> Then<Self, Fut, Fn>
+    where
+        Fn: FnOnce(Self::Output, &mut A, &mut A::Context) -> Fut,
+        Fut: IntoActorFuture<A>,
+        Self: Sized,
+    {
+        then::new(self, f)
+    }
+
+    /// Add timeout to futures chain.
+    ///
+    /// `Err(())` returned as a timeout error.
+    fn timeout(self, timeout: Duration) -> Timeout<Self>
+    where
+        Self: Sized,
+    {
+        Timeout::new(self, timeout)
+    }
+}
+
+impl<F, A> ActorFutureExt<A> for F
+where
+    F: ActorFuture<A>,
+    A: Actor,
+{
+}
+
+/// Type alias for a pinned box ActorFuture trait object.
+pub type LocalBoxActorFuture<A, I> = Pin<Box<dyn ActorFuture<A, Output = I>>>;
+
+/// Class of types which can be converted into an actor future.
+///
+/// This trait is very similar to the `IntoIterator` trait and is intended to be
+/// used in a very similar fashion.
+pub trait IntoActorFuture<A: Actor> {
+    /// The future that this type can be converted into.
+    type Future: ActorFuture<A, Output = Self::Output>;
+
+    /// The item that the future may resolve with.
+    type Output;
+
+    /// Consumes this object and produces a future.
+    fn into_future(self) -> Self::Future;
+}
+
+impl<F, A> IntoActorFuture<A> for F
+where
+    F: ActorFuture<A> + Sized,
+    A: Actor,
+{
+    type Future = F;
+    type Output = F::Output;
+
+    fn into_future(self) -> F {
+        self
+    }
+}
+
+impl<F, A> ActorFuture<A> for Box<F>
+where
+    F: ActorFuture<A> + Unpin + ?Sized,
+    A: Actor,
+{
+    type Output = F::Output;
+
+    fn poll(
+        mut self: Pin<&mut Self>,
+        srv: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Self::Output> {
+        Pin::new(&mut **self.as_mut()).poll(srv, ctx, task)
+    }
+}
+
+impl<P, A> ActorFuture<A> for Pin<P>
+where
+    P: Unpin + DerefMut,
+    <P as Deref>::Target: ActorFuture<A>,
+    A: Actor,
+{
+    type Output = <<P as Deref>::Target as ActorFuture<A>>::Output;
+
+    fn poll(
+        self: Pin<&mut Self>,
+        srv: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Self::Output> {
+        Pin::get_mut(self).as_mut().poll(srv, ctx, task)
+    }
+}
+
+/// Helper trait that allows conversion of normal future into `ActorFuture`
+pub trait WrapFuture<A>
+where
+    A: Actor,
+{
+    /// The future that this type can be converted into.
+    type Future: ActorFuture<A, Output = Self::Output>;
+
+    /// The item that the future may resolve with.
+    type Output;
+
+    #[deprecated(since = "0.11.0", note = "Please use WrapFuture::into_actor")]
+    #[doc(hidden)]
+    fn actfuture(self) -> Self::Future;
+
+    /// Convert normal future to a ActorFuture
+    fn into_actor(self, a: &A) -> Self::Future;
+}
+
+impl<F: Future, A: Actor> WrapFuture<A> for F {
+    type Future = FutureWrap<F, A>;
+    type Output = F::Output;
+
+    #[doc(hidden)]
+    fn actfuture(self) -> Self::Future {
+        wrap_future(self)
+    }
+
+    fn into_actor(self, _: &A) -> Self::Future {
+        wrap_future(self)
+    }
+}
+
+pin_project! {
+    pub struct FutureWrap<F, A>
+    where
+        F: Future,
+        A: Actor,
+    {
+        #[pin]
+        fut: F,
+        _act: PhantomData<A>
+    }
+}
+
+/// Converts normal future into `ActorFuture`, allowing its processing to
+/// use the actor's state.
+///
+/// See the documentation for [ActorFuture](trait.ActorFuture.html) for a practical example involving both
+/// `wrap_future` and `ActorFuture`
+pub fn wrap_future<F, A>(f: F) -> FutureWrap<F, A>
+where
+    F: Future,
+    A: Actor,
+{
+    FutureWrap {
+        fut: f,
+        _act: PhantomData,
+    }
+}
+
+impl<F, A> ActorFuture<A> for FutureWrap<F, A>
+where
+    F: Future,
+    A: Actor,
+{
+    type Output = F::Output;
+
+    fn poll(
+        self: Pin<&mut Self>,
+        _: &mut A,
+        _: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Self::Output> {
+        self.project().fut.poll(task)
+    }
+}
diff --git a/actix/src/fut/result.rs b/actix/src/fut/future/result.rs
similarity index 77%
rename from actix/src/fut/result.rs
rename to actix/src/fut/future/result.rs
index 7ba9ee67e..eecad7cb7 100644
--- a/actix/src/fut/result.rs
+++ b/actix/src/fut/future/result.rs
@@ -1,13 +1,20 @@
 //! Definition of the `Result` (immediately finished) combinator
 
-use super::ready_fut::{ready, Ready};
+use std::future::Future;
+use std::pin::Pin;
+use std::task;
+use std::task::Poll;
+
+use crate::actor::Actor;
+use crate::fut::ActorFuture;
+
+// TODO: remove re-export and encourage direct usage of std and/or futures crate types.
+pub use futures_util::future::{ready, Ready};
 
 /// A future representing a value that is immediately ready.
 ///
 /// Created by the `result` function.
 #[must_use = "futures do nothing unless polled"]
-// TODO: rename this to `Result` on the next major version
-// TODO: use std::future::Ready when MSRV surpass 1.48
 pub type FutureResult<T, E> = Ready<Result<T, E>>;
 
 /// Creates a new "leaf future" which will resolve with the given result.
@@ -76,3 +83,20 @@ pub fn ok<T, E>(t: T) -> FutureResult<T, E> {
 pub fn err<T, E>(e: E) -> FutureResult<T, E> {
     ready(Err(e))
 }
+
+impl<T, A> ActorFuture<A> for Ready<T>
+where
+    A: Actor,
+{
+    type Output = T;
+
+    #[inline]
+    fn poll(
+        self: Pin<&mut Self>,
+        _: &mut A,
+        _: &mut A::Context,
+        cx: &mut task::Context<'_>,
+    ) -> Poll<Self::Output> {
+        Future::poll(self, cx)
+    }
+}
diff --git a/actix/src/fut/future/then.rs b/actix/src/fut/future/then.rs
new file mode 100644
index 000000000..72827db21
--- /dev/null
+++ b/actix/src/fut/future/then.rs
@@ -0,0 +1,75 @@
+use std::pin::Pin;
+use std::task::{self, Poll};
+
+use futures_core::ready;
+use pin_project_lite::pin_project;
+
+use crate::actor::Actor;
+use crate::fut::{ActorFuture, IntoActorFuture};
+
+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.
+    #[project = ThenProj]
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless polled"]
+    pub enum Then<A, B, Fn> {
+        First {
+            #[pin]
+            fut1: A,
+            data: Option<Fn>,
+        },
+        Second {
+            #[pin]
+            fut2: B
+        },
+        Empty,
+    }
+}
+
+pub(super) fn new<A, B, Fn, Act>(future: A, f: Fn) -> Then<A, B, Fn>
+where
+    A: ActorFuture<Act>,
+    B: IntoActorFuture<Act>,
+    Act: Actor,
+{
+    Then::First {
+        fut1: future,
+        data: Some(f),
+    }
+}
+
+impl<A, B, Fn, Act> ActorFuture<Act> for Then<A, B::Future, Fn>
+where
+    A: ActorFuture<Act>,
+    B: IntoActorFuture<Act>,
+    Fn: FnOnce(A::Output, &mut Act, &mut Act::Context) -> B,
+    Act: Actor,
+{
+    type Output = B::Output;
+
+    fn poll(
+        mut self: Pin<&mut Self>,
+        act: &mut Act,
+        ctx: &mut Act::Context,
+        task: &mut task::Context<'_>,
+    ) -> Poll<B::Output> {
+        match self.as_mut().project() {
+            ThenProj::First { fut1, data } => {
+                let output = ready!(fut1.poll(act, ctx, task));
+                let data = data.take().unwrap();
+                let fut2 = data(output, act, ctx).into_future();
+                self.set(Then::Second { fut2 });
+                self.poll(act, ctx, task)
+            }
+            ThenProj::Second { fut2 } => {
+                let res = ready!(fut2.poll(act, ctx, task));
+                self.set(Then::Empty);
+                Poll::Ready(res)
+            }
+            ThenProj::Empty => panic!("ActorFuture polled after finish"),
+        }
+    }
+}
diff --git a/actix/src/fut/timeout.rs b/actix/src/fut/future/timeout.rs
similarity index 74%
rename from actix/src/fut/timeout.rs
rename to actix/src/fut/future/timeout.rs
index b5a183892..9e48d81bb 100644
--- a/actix/src/fut/timeout.rs
+++ b/actix/src/fut/future/timeout.rs
@@ -24,10 +24,12 @@ pin_project! {
     }
 }
 
-pub fn new<F>(future: F, timeout: Duration) -> Timeout<F> {
-    Timeout {
-        fut: future,
-        timeout: sleep(timeout),
+impl<F> Timeout<F> {
+    pub(super) fn new(future: F, timeout: Duration) -> Self {
+        Self {
+            fut: future,
+            timeout: sleep(timeout),
+        }
     }
 }
 
@@ -45,11 +47,9 @@ where
         task: &mut Context<'_>,
     ) -> Poll<Self::Output> {
         let this = self.project();
-
-        if this.timeout.poll(task).is_ready() {
-            return Poll::Ready(Err(()));
+        match this.fut.poll(act, ctx, task) {
+            Poll::Ready(res) => Poll::Ready(Ok(res)),
+            Poll::Pending => this.timeout.poll(task).map(Err),
         }
-
-        this.fut.poll(act, ctx, task).map(Ok)
     }
 }
diff --git a/actix/src/fut/helpers.rs b/actix/src/fut/helpers.rs
deleted file mode 100644
index f940452e6..000000000
--- a/actix/src/fut/helpers.rs
+++ /dev/null
@@ -1,57 +0,0 @@
-use std::future::Future;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-
-use futures_core::stream::Stream;
-use pin_project_lite::pin_project;
-
-/// Helper trait that adds the helper method `finish()` to stream objects.
-#[doc(hidden)]
-pub trait FinishStream: Sized {
-    fn finish(self) -> Finish<Self>;
-}
-
-impl<S: Stream> FinishStream for S {
-    /// A combinator used to convert a stream into a future; the
-    /// future resolves when the stream completes.
-    fn finish(self) -> Finish<S> {
-        Finish::new(self)
-    }
-}
-
-pin_project! {
-    /// A combinator used to convert a stream into a future; the future
-    /// resolves when the stream completes.
-    ///
-    /// This structure is produced by the `Stream::finish` method.
-    #[derive(Debug)]
-    #[must_use = "streams do nothing unless polled"]
-    pub struct Finish<S> {
-        #[pin]
-        stream: S
-    }
-}
-
-impl<S> Finish<S> {
-    pub fn new(stream: S) -> Finish<S> {
-        Finish { stream }
-    }
-}
-
-impl<S> Future for Finish<S>
-where
-    S: Stream,
-{
-    type Output = ();
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        loop {
-            let this = self.as_mut().project();
-            match this.stream.poll_next(cx) {
-                Poll::Pending => return Poll::Pending,
-                Poll::Ready(None) => return Poll::Ready(()),
-                Poll::Ready(Some(_)) => (),
-            };
-        }
-    }
-}
diff --git a/actix/src/fut/map.rs b/actix/src/fut/map.rs
deleted file mode 100644
index 9b7822648..000000000
--- a/actix/src/fut/map.rs
+++ /dev/null
@@ -1,47 +0,0 @@
-use std::pin::Pin;
-use std::task::{Context, Poll};
-
-use pin_project_lite::pin_project;
-
-use crate::actor::Actor;
-use crate::fut::ActorFuture;
-
-pin_project! {
-    /// Future for the `map` combinator, changing the type of a future.
-    ///
-    /// This is created by the `ActorFuture::map` method
-    #[derive(Debug)]
-    #[must_use = "futures do nothing unless polled"]
-    pub struct Map<F, Fn> {
-        #[pin]
-        future: F,
-        f: Option<Fn>,
-    }
-}
-
-pub fn new<F, Fn>(future: F, f: Fn) -> Map<F, Fn> {
-    Map { future, f: Some(f) }
-}
-
-impl<U, F, A, Fn> ActorFuture<A> for Map<F, Fn>
-where
-    F: ActorFuture<A>,
-    A: Actor,
-    Fn: FnOnce(F::Output, &mut A, &mut A::Context) -> U,
-{
-    type Output = U;
-
-    fn poll(
-        self: Pin<&mut Self>,
-        act: &mut A,
-        ctx: &mut A::Context,
-        task: &mut Context<'_>,
-    ) -> Poll<Self::Output> {
-        let this = self.project();
-        let e = match this.future.poll(act, ctx, task) {
-            Poll::Pending => return Poll::Pending,
-            Poll::Ready(e) => e,
-        };
-        Poll::Ready(this.f.take().expect("cannot poll Map twice")(e, act, ctx))
-    }
-}
diff --git a/actix/src/fut/mod.rs b/actix/src/fut/mod.rs
index 69387ccee..7e1fbc2af 100644
--- a/actix/src/fut/mod.rs
+++ b/actix/src/fut/mod.rs
@@ -1,448 +1,10 @@
-//! Custom `Future` implementation with `Actix` support
+//! Custom `Future` and `Stream` implementation with `Actix` support
 
-use std::future::Future;
-use std::marker::PhantomData;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use std::time::Duration;
+pub mod future;
+pub mod stream;
 
-use futures_core::stream::Stream;
-use pin_project_lite::pin_project;
-
-mod chain;
-mod either;
-mod helpers;
-mod map;
-mod ready_fut;
-mod result;
-mod stream_finish;
-mod stream_fold;
-mod stream_map;
-mod stream_then;
-mod stream_timeout;
-mod then;
-mod timeout;
-
-pub use self::either::Either;
-pub use self::helpers::{Finish, FinishStream};
-pub use self::map::Map;
-pub use self::ready_fut::{ready, Ready};
-pub use self::result::{err, ok, result, FutureResult};
-pub use self::stream_finish::StreamFinish;
-pub use self::stream_fold::StreamFold;
-pub use self::stream_map::StreamMap;
-pub use self::stream_then::StreamThen;
-pub use self::stream_timeout::StreamTimeout;
-pub use self::then::Then;
-pub use self::timeout::Timeout;
-
-use crate::actor::Actor;
-
-/// Trait for types which are a placeholder of a value that may become
-/// available at some later point in time.
-///
-/// `ActorFuture` is very similar to a regular `Future`, only with subsequent combinator closures accepting the actor and its context, in addition to the result.
-///
-/// `ActorFuture` allows for use cases where future processing requires access to the actor or its context.
-///
-/// Here is an example of a handler on a single actor, deferring work to another actor, and
-/// then updating the initiating actor's state:
-///
-/// ```rust,no_run
-/// use actix::prelude::*;
-///
-/// // The response type returned by the actor future
-/// type OriginalActorResponse = ();
-/// // The error type returned by the actor future
-/// type MessageError = ();
-/// // This is the needed result for the DeferredWork message
-/// // It's a result that combine both Response and Error from the future response.
-/// type DeferredWorkResult = Result<OriginalActorResponse, MessageError>;
-/// #
-/// # struct ActorState {}
-/// #
-/// # impl ActorState {
-/// #    fn update_from(&mut self, _result: ()) {}
-/// # }
-/// #
-/// # struct OtherActor {}
-/// #
-/// # impl Actor for OtherActor {
-/// #    type Context = Context<Self>;
-/// # }
-/// #
-/// # impl Handler<OtherMessage> for OtherActor {
-/// #    type Result = ();
-/// #
-/// #    fn handle(&mut self, _msg: OtherMessage, _ctx: &mut Context<Self>) -> Self::Result {
-/// #    }
-/// # }
-/// #
-/// # struct OriginalActor{
-/// #     other_actor: Addr<OtherActor>,
-/// #     inner_state: ActorState
-/// # }
-/// #
-/// # impl Actor for OriginalActor{
-/// #     type Context = Context<Self>;
-/// # }
-/// #
-/// # #[derive(Message)]
-/// # #[rtype(result = "Result<(), MessageError>")]
-/// # struct DeferredWork{}
-/// #
-/// # #[derive(Message)]
-/// # #[rtype(result = "()")]
-/// # struct OtherMessage{}
-///
-/// impl Handler<DeferredWork> for OriginalActor {
-///     // Notice the `Response` is an `ActorFuture`-ized version of `Self::Message::Result`.
-///     type Result = ResponseActFuture<Self, Result<OriginalActorResponse, MessageError>>;
-///
-///     fn handle(&mut self, _msg: DeferredWork, _ctx: &mut Context<Self>) -> Self::Result {
-///         // this creates a `Future` representing the `.send` and subsequent `Result` from
-///         // `other_actor`
-///         let send_to_other = self.other_actor
-///             .send(OtherMessage {});
-///
-///         // Wrap that `Future` so subsequent chained handlers can access
-///         // the `actor` (`self` in the  synchronous code) as well as the context.
-///         let send_to_other = actix::fut::wrap_future::<_, Self>(send_to_other);
-///
-///         // once the wrapped future resolves, update this actor's state
-///         let update_self = send_to_other.map(|result, actor, _ctx| {
-///             // Actor's state updated here
-///             match result {
-///                 Ok(v) => {
-///                     actor.inner_state.update_from(v);
-///                     Ok(())
-///                 },
-///                 // Failed to send message to other_actor
-///                 Err(_e) => Err(()),
-///             }
-///         });
-///
-///         // return the wrapped future
-///         Box::pin(update_self)
-///     }
-/// }
-///
-/// ```
-///
-/// See also [into_actor](trait.WrapFuture.html#tymethod.into_actor), which provides future conversion using trait
-pub trait ActorFuture<A: Actor> {
-    /// The type of value that this future will resolved with if it is
-    /// successful.
-    type Output;
-
-    fn poll(
-        self: Pin<&mut Self>,
-        srv: &mut A,
-        ctx: &mut A::Context,
-        task: &mut Context<'_>,
-    ) -> Poll<Self::Output>;
-
-    /// Map this future's result to a different type, returning a new future of
-    /// the resulting type.
-    fn map<F, U>(self, f: F) -> Map<Self, F>
-    where
-        F: FnOnce(Self::Output, &mut A, &mut A::Context) -> U,
-        Self: Sized,
-    {
-        map::new(self, f)
-    }
-
-    /// Chain on a computation for when a future finished, passing the result of
-    /// the future to the provided closure `f`.
-    fn then<F, B>(self, f: F) -> Then<Self, B, F, A>
-    where
-        F: FnOnce(Self::Output, &mut A, &mut A::Context) -> B,
-        B: IntoActorFuture<A>,
-        Self: Sized,
-    {
-        then::new(self, f)
-    }
-
-    /// Add timeout to futures chain.
-    ///
-    /// `err` value get returned as a timeout error.
-    fn timeout(self, timeout: Duration) -> Timeout<Self>
-    where
-        Self: Sized,
-    {
-        timeout::new(self, timeout)
-    }
-}
-
-/// A stream of values, not all of which may have been produced yet.
-///
-/// This is similar to `futures_util::stream::Stream` trait, except it works with `Actor`
-pub trait ActorStream<A: Actor> {
-    /// The type of item this stream will yield on success.
-    type Item;
-
-    fn poll_next(
-        self: Pin<&mut Self>,
-        srv: &mut A,
-        ctx: &mut A::Context,
-        task: &mut Context<'_>,
-    ) -> Poll<Option<Self::Item>>;
-
-    /// Converts a stream of type `T` to a stream of type `U`.
-    fn map<U, F>(self, f: F) -> StreamMap<Self, F>
-    where
-        F: FnMut(Self::Item, &mut A, &mut A::Context) -> U,
-        Self: Sized,
-    {
-        stream_map::new(self, f)
-    }
-
-    /// Chain on a computation for when a value is ready, passing the resulting
-    /// item to the provided closure `f`.
-    fn then<F, U>(self, f: F) -> StreamThen<Self, F, U::Future>
-    where
-        F: FnMut(Self::Item, &mut A, &mut A::Context) -> U,
-        U: IntoActorFuture<A>,
-        Self: Sized,
-    {
-        stream_then::new(self, f)
-    }
-
-    /// Execute an accumulating computation over a stream, collecting all the
-    /// values into one final result.
-    fn fold<F, T, Fut>(self, init: T, f: F) -> StreamFold<Self, F, Fut::Future, T>
-    where
-        F: FnMut(T, Self::Item, &mut A, &mut A::Context) -> Fut,
-        Fut: IntoActorFuture<A, Output = T>,
-        Self: Sized,
-    {
-        stream_fold::new(self, f, init)
-    }
-
-    /// Add timeout to stream.
-    ///
-    /// `err` value get returned as a timeout error.
-    fn timeout(self, timeout: Duration) -> StreamTimeout<Self>
-    where
-        Self: Sized,
-    {
-        stream_timeout::new(self, timeout)
-    }
-
-    /// Converts a stream to a future that resolves when stream finishes.
-    fn finish(self) -> StreamFinish<Self>
-    where
-        Self: Sized,
-    {
-        stream_finish::new(self)
-    }
-}
-
-/// Class of types which can be converted into an actor future.
-///
-/// This trait is very similar to the `IntoIterator` trait and is intended to be
-/// used in a very similar fashion.
-pub trait IntoActorFuture<A: Actor> {
-    /// The future that this type can be converted into.
-    type Future: ActorFuture<A, Output = Self::Output>;
-
-    /// The item that the future may resolve with.
-    type Output;
-
-    /// Consumes this object and produces a future.
-    fn into_future(self) -> Self::Future;
-}
-
-impl<F, A> IntoActorFuture<A> for F
-where
-    F: ActorFuture<A> + Sized,
-    A: Actor,
-{
-    type Future = F;
-    type Output = F::Output;
-
-    fn into_future(self) -> F {
-        self
-    }
-}
-
-impl<F, A> ActorFuture<A> for Box<F>
-where
-    F: ActorFuture<A> + Unpin + ?Sized,
-    A: Actor,
-{
-    type Output = F::Output;
-
-    fn poll(
-        mut self: Pin<&mut Self>,
-        srv: &mut A,
-        ctx: &mut A::Context,
-        task: &mut Context<'_>,
-    ) -> Poll<Self::Output> {
-        Pin::new(&mut **self.as_mut()).poll(srv, ctx, task)
-    }
-}
-
-impl<P, A> ActorFuture<A> for Pin<P>
-where
-    P: Unpin + std::ops::DerefMut,
-    <P as std::ops::Deref>::Target: ActorFuture<A>,
-    A: Actor,
-{
-    type Output = <<P as std::ops::Deref>::Target as ActorFuture<A>>::Output;
-
-    fn poll(
-        self: Pin<&mut Self>,
-        srv: &mut A,
-        ctx: &mut A::Context,
-        task: &mut Context<'_>,
-    ) -> Poll<Self::Output> {
-        Pin::get_mut(self).as_mut().poll(srv, ctx, task)
-    }
-}
-
-/// Helper trait that allows conversion of normal future into `ActorFuture`
-pub trait WrapFuture<A>
-where
-    A: Actor,
-{
-    /// The future that this type can be converted into.
-    type Future: ActorFuture<A, Output = Self::Output>;
-
-    /// The item that the future may resolve with.
-    type Output;
-
-    #[doc(hidden)]
-    fn actfuture(self) -> Self::Future;
-
-    /// Convert normal future to a ActorFuture
-    fn into_actor(self, a: &A) -> Self::Future;
-}
-
-impl<F: Future, A: Actor> WrapFuture<A> for F {
-    type Future = FutureWrap<F, A>;
-    type Output = F::Output;
-
-    #[doc(hidden)]
-    fn actfuture(self) -> Self::Future {
-        wrap_future(self)
-    }
-
-    fn into_actor(self, _: &A) -> Self::Future {
-        wrap_future(self)
-    }
-}
-
-pin_project! {
-    pub struct FutureWrap<F, A>
-    where
-        F: Future,
-    {
-        #[pin]
-        fut: F,
-        _act: PhantomData<A>
-    }
-}
-
-/// Converts normal future into `ActorFuture`, allowing its processing to
-/// use the actor's state.
-///
-/// See the documentation for [ActorFuture](trait.ActorFuture.html) for a practical example involving both
-/// `wrap_future` and `ActorFuture`
-pub fn wrap_future<F, A>(f: F) -> FutureWrap<F, A>
-where
-    F: Future,
-{
-    FutureWrap {
-        fut: f,
-        _act: PhantomData,
-    }
-}
-
-impl<F, A> ActorFuture<A> for FutureWrap<F, A>
-where
-    F: Future,
-    A: Actor,
-{
-    type Output = F::Output;
-
-    fn poll(
-        self: Pin<&mut Self>,
-        _: &mut A,
-        _: &mut A::Context,
-        task: &mut Context<'_>,
-    ) -> Poll<Self::Output> {
-        self.project().fut.poll(task)
-    }
-}
-
-/// Helper trait that allows conversion of normal stream into `ActorStream`
-pub trait WrapStream<A>
-where
-    A: Actor,
-{
-    /// The stream that this type can be converted into.
-    type Stream: ActorStream<A, Item = Self::Item>;
-
-    /// The item that the future may resolve with.
-    type Item;
-
-    #[doc(hidden)]
-    fn actstream(self) -> Self::Stream;
-
-    /// Convert normal stream to a ActorStream
-    fn into_actor(self, a: &A) -> Self::Stream;
-}
-
-impl<S: Stream, A: Actor> WrapStream<A> for S {
-    type Stream = StreamWrap<S, A>;
-    type Item = S::Item;
-
-    #[doc(hidden)]
-    fn actstream(self) -> Self::Stream {
-        wrap_stream(self)
-    }
-
-    fn into_actor(self, _: &A) -> Self::Stream {
-        wrap_stream(self)
-    }
-}
-
-pin_project! {
-    pub struct StreamWrap<S, A>
-    where
-        S: Stream,
-    {
-        #[pin]
-        st: S,
-        _act: PhantomData<A>
-    }
-}
-
-/// Converts normal stream into `ActorStream`
-pub fn wrap_stream<S, A>(s: S) -> StreamWrap<S, A>
-where
-    S: Stream,
-{
-    StreamWrap {
-        st: s,
-        _act: PhantomData,
-    }
-}
-
-impl<S, A> ActorStream<A> for StreamWrap<S, A>
-where
-    S: Stream,
-    A: Actor,
-{
-    type Item = S::Item;
-
-    fn poll_next(
-        self: Pin<&mut Self>,
-        _: &mut A,
-        _: &mut A::Context,
-        task: &mut Context<'_>,
-    ) -> Poll<Option<Self::Item>> {
-        self.project().st.poll_next(task)
-    }
-}
+pub use self::future::{
+    result::*, wrap_future, ActorFuture, ActorFutureExt, Either, IntoActorFuture,
+    LocalBoxActorFuture, WrapFuture,
+};
+pub use self::stream::{wrap_stream, ActorStream, ActorStreamExt, WrapStream};
diff --git a/actix/src/fut/ready_fut.rs b/actix/src/fut/ready_fut.rs
deleted file mode 100644
index b1cced32a..000000000
--- a/actix/src/fut/ready_fut.rs
+++ /dev/null
@@ -1,27 +0,0 @@
-//! Definition of the `Result` (immediately finished) combinator
-use std::future::Future;
-use std::pin::Pin;
-use std::task;
-use std::task::Poll;
-
-use crate::actor::Actor;
-use crate::fut::ActorFuture;
-
-pub use futures_util::future::{ready, Ready};
-
-impl<T, A> ActorFuture<A> for Ready<T>
-where
-    A: Actor,
-{
-    type Output = T;
-
-    #[inline]
-    fn poll(
-        self: Pin<&mut Self>,
-        _: &mut A,
-        _: &mut A::Context,
-        cx: &mut task::Context<'_>,
-    ) -> Poll<Self::Output> {
-        Future::poll(self, cx)
-    }
-}
diff --git a/actix/src/fut/stream/collect.rs b/actix/src/fut/stream/collect.rs
new file mode 100644
index 000000000..4cbc89e66
--- /dev/null
+++ b/actix/src/fut/stream/collect.rs
@@ -0,0 +1,60 @@
+use std::{
+    mem,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use futures_core::ready;
+use pin_project_lite::pin_project;
+
+use super::ActorStream;
+
+use crate::actor::Actor;
+use crate::fut::future::ActorFuture;
+
+pin_project! {
+    /// Future for the [`collect`](super::ActorStreamExt::collect) method.
+    #[derive(Debug)]
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct Collect<S, C> {
+        #[pin]
+        stream: S,
+        collection: C,
+    }
+}
+
+impl<S, C> Collect<S, C>
+where
+    C: Default,
+{
+    pub(super) fn new(stream: S) -> Self {
+        Self {
+            stream,
+            collection: Default::default(),
+        }
+    }
+}
+
+impl<S, A, C> ActorFuture<A> for Collect<S, C>
+where
+    S: ActorStream<A>,
+    A: Actor,
+    C: Default + Extend<S::Item>,
+{
+    type Output = C;
+
+    fn poll(
+        mut self: Pin<&mut Self>,
+        act: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Self::Output> {
+        let mut this = self.as_mut().project();
+        loop {
+            match ready!(this.stream.as_mut().poll_next(act, ctx, task)) {
+                Some(e) => this.collection.extend(Some(e)),
+                None => return Poll::Ready(mem::take(this.collection)),
+            }
+        }
+    }
+}
diff --git a/actix/src/fut/stream/finish.rs b/actix/src/fut/stream/finish.rs
new file mode 100644
index 000000000..35985b656
--- /dev/null
+++ b/actix/src/fut/stream/finish.rs
@@ -0,0 +1,43 @@
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use futures_core::ready;
+use pin_project_lite::pin_project;
+
+use crate::actor::Actor;
+use crate::fut::{ActorFuture, ActorStream};
+
+pin_project! {
+    /// Future for the [`finish`](super::ActorStreamExt::finish) method.
+    #[derive(Debug)]
+    #[must_use = "streams do nothing unless polled"]
+    pub struct Finish<S> {
+        #[pin]
+        pub(crate) stream: S
+    }
+}
+
+impl<S> Finish<S> {
+    pub fn new(stream: S) -> Finish<S> {
+        Finish { stream }
+    }
+}
+
+impl<S, A> ActorFuture<A> for Finish<S>
+where
+    S: ActorStream<A>,
+    A: Actor,
+{
+    type Output = ();
+
+    fn poll(
+        mut self: Pin<&mut Self>,
+        act: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<()> {
+        let mut this = self.as_mut().project();
+        while ready!(this.stream.as_mut().poll_next(act, ctx, task)).is_some() {}
+        Poll::Ready(())
+    }
+}
diff --git a/actix/src/fut/stream/fold.rs b/actix/src/fut/stream/fold.rs
new file mode 100644
index 000000000..d3950abfd
--- /dev/null
+++ b/actix/src/fut/stream/fold.rs
@@ -0,0 +1,80 @@
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use futures_core::ready;
+use pin_project_lite::pin_project;
+
+use crate::actor::Actor;
+use crate::fut::{ActorFuture, ActorStream, IntoActorFuture};
+
+pin_project! {
+    /// Stream for the [`fold`](super::ActorStreamExt::fold) method.
+    #[derive(Debug)]
+    #[must_use = "streams do nothing unless polled"]
+    pub struct Fold<S, F, Fut, T> {
+        #[pin]
+        stream: S,
+        f: F,
+        accum: Option<T>,
+        #[pin]
+        future: Option<Fut>,
+    }
+}
+
+pub(super) fn new<S, A, F, Fut>(
+    stream: S,
+    f: F,
+    t: Fut::Output,
+) -> Fold<S, F, Fut::Future, Fut::Output>
+where
+    S: ActorStream<A>,
+    A: Actor,
+    F: FnMut(Fut::Output, S::Item, &mut A, &mut A::Context) -> Fut,
+    Fut: IntoActorFuture<A>,
+{
+    Fold {
+        stream,
+        f,
+        accum: Some(t),
+        future: None,
+    }
+}
+
+impl<S, A, F, Fut, T> ActorFuture<A> for Fold<S, F, Fut::Future, T>
+where
+    S: ActorStream<A>,
+    A: Actor,
+    F: FnMut(T, S::Item, &mut A, &mut A::Context) -> Fut,
+    Fut: IntoActorFuture<A, Output = T>,
+    Fut::Future: ActorFuture<A>,
+{
+    type Output = T;
+
+    fn poll(
+        self: Pin<&mut Self>,
+        act: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<T> {
+        let mut this = self.project();
+        Poll::Ready(loop {
+            if let Some(fut) = this.future.as_mut().as_pin_mut() {
+                // we're currently processing a future to produce a new accum value
+                *this.accum = Some(ready!(fut.poll(act, ctx, task)));
+                this.future.set(None);
+            } else if this.accum.is_some() {
+                // we're waiting on a new item from the stream
+                let res = ready!(this.stream.as_mut().poll_next(act, ctx, task));
+                let a = this.accum.take().unwrap();
+                if let Some(item) = res {
+                    this.future
+                        .set(Some((this.f)(a, item, act, ctx).into_future()));
+                } else {
+                    break a;
+                }
+            } else {
+                panic!("Fold polled after completion")
+            }
+        })
+    }
+}
diff --git a/actix/src/fut/stream/map.rs b/actix/src/fut/stream/map.rs
new file mode 100644
index 000000000..3285309d6
--- /dev/null
+++ b/actix/src/fut/stream/map.rs
@@ -0,0 +1,48 @@
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use futures_core::ready;
+use pin_project_lite::pin_project;
+
+use crate::actor::Actor;
+use crate::fut::ActorStream;
+
+pin_project! {
+    /// Stream for the [`map`](super::ActorStreamExt::map) method.
+    #[derive(Debug)]
+    #[must_use = "streams do nothing unless polled"]
+    pub struct Map<S, F> {
+        #[pin]
+        stream: S,
+        f: F,
+    }
+}
+
+pub(super) fn new<S, A, F, U>(stream: S, f: F) -> Map<S, F>
+where
+    F: FnMut(S::Item, &mut A, &mut A::Context) -> U,
+    S: ActorStream<A>,
+    A: Actor,
+{
+    Map { stream, f }
+}
+
+impl<S, A, F, U> ActorStream<A> for Map<S, F>
+where
+    S: ActorStream<A>,
+    A: Actor,
+    F: FnMut(S::Item, &mut A, &mut A::Context) -> U,
+{
+    type Item = U;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        act: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Option<U>> {
+        let mut this = self.project();
+        let res = ready!(this.stream.as_mut().poll_next(act, ctx, task));
+        Poll::Ready(res.map(|x| (this.f)(x, act, ctx)))
+    }
+}
diff --git a/actix/src/fut/stream/mod.rs b/actix/src/fut/stream/mod.rs
new file mode 100644
index 000000000..cb21064dc
--- /dev/null
+++ b/actix/src/fut/stream/mod.rs
@@ -0,0 +1,247 @@
+use std::{
+    marker::PhantomData,
+    pin::Pin,
+    task::{Context, Poll},
+    time::Duration,
+};
+
+use futures_core::stream::Stream;
+use pin_project_lite::pin_project;
+
+pub use collect::Collect;
+pub use finish::Finish;
+pub use fold::Fold;
+pub use map::Map;
+pub use skip_while::SkipWhile;
+pub use take_while::TakeWhile;
+pub use then::Then;
+pub use timeout::Timeout;
+
+use crate::actor::Actor;
+
+use super::future::IntoActorFuture;
+
+mod collect;
+mod finish;
+mod fold;
+mod map;
+mod skip_while;
+mod take_while;
+mod then;
+mod timeout;
+
+/// A stream of values, not all of which may have been produced yet.
+///
+/// This is similar to `futures_util::stream::Stream` trait, except it works with `Actor`
+pub trait ActorStream<A: Actor> {
+    /// The type of item this stream will yield on success.
+    type Item;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        srv: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Option<Self::Item>>;
+}
+
+pub trait ActorStreamExt<A: Actor>: ActorStream<A> {
+    /// Maps this stream's items to a different type, returning a new stream of
+    /// the resulting type.
+    ///
+    /// The provided closure is executed over all elements of this stream as
+    /// they are made available. It is executed inline with calls to
+    /// [`poll_next`](ActorStream::poll_next).
+    ///
+    /// Note that this function consumes the stream passed into it and returns a
+    /// wrapped version of it, similar to the existing `map` methods in the
+    /// standard library.
+    fn map<Fn, U>(self, f: Fn) -> Map<Self, Fn>
+    where
+        Fn: FnMut(Self::Item, &mut A, &mut A::Context) -> U,
+        Self: Sized,
+    {
+        map::new(self, f)
+    }
+
+    /// Computes from this stream's items new items of a different type using
+    /// an asynchronous closure.
+    ///
+    /// The provided closure `f` will be called with an `Item` once a value is
+    /// ready, it returns a future which will then be run to completion
+    /// to produce the next value on this stream.
+    ///
+    /// Note that this function consumes the stream passed into it and returns a
+    /// wrapped version of it.
+    fn then<Fn, Fut>(self, f: Fn) -> Then<Self, Fn, Fut::Future>
+    where
+        Fn: FnMut(Self::Item, &mut A, &mut A::Context) -> Fut,
+        Fut: IntoActorFuture<A>,
+        Self: Sized,
+    {
+        then::new(self, f)
+    }
+
+    /// Execute an accumulating asynchronous computation over a stream,
+    /// collecting all the values into one final result.
+    ///
+    /// This combinator will accumulate all values returned by this stream
+    /// according to the closure provided. The initial state is also provided to
+    /// this method and then is returned again by each execution of the closure.
+    /// Once the entire stream has been exhausted the returned future will
+    /// resolve to this value.
+    fn fold<Fn, Fut>(self, init: Fut::Output, f: Fn) -> Fold<Self, Fn, Fut::Future, Fut::Output>
+    where
+        Fn: FnMut(Fut::Output, Self::Item, &mut A, &mut A::Context) -> Fut,
+        Fut: IntoActorFuture<A>,
+        Self: Sized,
+    {
+        fold::new(self, f, init)
+    }
+
+    /// Take elements from this stream while the provided asynchronous predicate
+    /// resolves to `true`.
+    ///
+    /// This function, like `Iterator::take_while`, will take elements from the
+    /// stream until the predicate `f` resolves to `false`. Once one element
+    /// returns `false`, it will always return that the stream is done.
+    fn take_while<Fn, Fut>(self, f: Fn) -> TakeWhile<Self, Self::Item, Fut::Future, Fn>
+    where
+        Fn: FnMut(&Self::Item, &mut A, &mut A::Context) -> Fut,
+        Fut: IntoActorFuture<A, Output = bool>,
+        Self: Sized,
+    {
+        take_while::new(self, f)
+    }
+
+    /// Skip elements on this stream while the provided asynchronous predicate
+    /// resolves to `true`.
+    ///
+    /// This function, like `Iterator::skip_while`, will skip elements on the
+    /// stream until the predicate `f` resolves to `false`. Once one element
+    /// returns `false`, all future elements will be returned from the underlying
+    /// stream.
+    fn skip_while<Fn, Fut>(self, f: Fn) -> SkipWhile<Self, Self::Item, Fn, Fut::Future>
+    where
+        Fn: FnMut(&Self::Item, &mut A, &mut A::Context) -> Fut,
+        Fut: IntoActorFuture<A, Output = bool>,
+        Self: Sized,
+    {
+        skip_while::new(self, f)
+    }
+
+    /// Add timeout to stream.
+    ///
+    /// `Err(())` returned as a timeout error.
+    fn timeout(self, timeout: Duration) -> Timeout<Self>
+    where
+        Self: Sized,
+    {
+        Timeout::new(self, timeout)
+    }
+
+    /// Transforms a stream into a collection, returning a
+    /// future representing the result of that computation.
+    ///
+    /// The returned future will be resolved when the stream terminates.
+    fn collect<C>(self) -> Collect<Self, C>
+    where
+        C: Default + Extend<Self::Item>,
+        Self: Sized,
+    {
+        Collect::new(self)
+    }
+
+    /// Transforms a stream to a future that resolves when stream finishes.
+    fn finish(self) -> Finish<Self>
+    where
+        Self: Sized,
+    {
+        Finish::new(self)
+    }
+}
+
+impl<A, S> ActorStreamExt<A> for S
+where
+    S: ActorStream<A>,
+    A: Actor,
+{
+}
+
+/// Helper trait that allows conversion of normal stream into `ActorStream`
+pub trait WrapStream<A>
+where
+    A: Actor,
+{
+    /// The stream that this type can be converted into.
+    type Stream: ActorStream<A, Item = Self::Item>;
+
+    /// The item that the future may resolve with.
+    type Item;
+
+    #[deprecated(since = "0.11.0", note = "Please use WrapStream::into_actor")]
+    #[doc(hidden)]
+    fn actstream(self) -> Self::Stream;
+
+    /// Convert normal stream to a ActorStream
+    fn into_actor(self, a: &A) -> Self::Stream;
+}
+
+impl<S, A> WrapStream<A> for S
+where
+    S: Stream,
+    A: Actor,
+{
+    type Stream = StreamWrap<S, A>;
+    type Item = S::Item;
+
+    #[doc(hidden)]
+    fn actstream(self) -> Self::Stream {
+        wrap_stream(self)
+    }
+
+    fn into_actor(self, _: &A) -> Self::Stream {
+        wrap_stream(self)
+    }
+}
+
+pin_project! {
+    pub struct StreamWrap<S, A>
+    where
+        S: Stream,
+        A: Actor
+    {
+        #[pin]
+        stream: S,
+        _act: PhantomData<A>
+    }
+}
+
+/// Converts normal stream into `ActorStream`
+pub fn wrap_stream<S, A>(stream: S) -> StreamWrap<S, A>
+where
+    S: Stream,
+    A: Actor,
+{
+    StreamWrap {
+        stream,
+        _act: PhantomData,
+    }
+}
+
+impl<S, A> ActorStream<A> for StreamWrap<S, A>
+where
+    S: Stream,
+    A: Actor,
+{
+    type Item = S::Item;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        _: &mut A,
+        _: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        self.project().stream.poll_next(task)
+    }
+}
diff --git a/actix/src/fut/stream/skip_while.rs b/actix/src/fut/stream/skip_while.rs
new file mode 100644
index 000000000..3425c0cfd
--- /dev/null
+++ b/actix/src/fut/stream/skip_while.rs
@@ -0,0 +1,80 @@
+use std::pin::Pin;
+use std::task::{self, Poll};
+
+use futures_core::ready;
+use pin_project_lite::pin_project;
+
+use crate::actor::Actor;
+use crate::fut::{ActorFuture, ActorStream, IntoActorFuture};
+
+pin_project! {
+    /// Stream for the [`skip_while`](super::ActorStreamExt::skip_while) method.
+    #[derive(Debug)]
+    #[must_use = "streams do nothing unless polled"]
+    pub struct SkipWhile<S, I, Fn, Fut> {
+        #[pin]
+        stream: S,
+        f: Fn,
+        #[pin]
+        pending_fut: Option<Fut>,
+        pending_item: Option<I>,
+        done_skipping: bool,
+    }
+}
+
+pub(super) fn new<S, A, Fn, Fut>(stream: S, f: Fn) -> SkipWhile<S, S::Item, Fn, Fut::Future>
+where
+    S: ActorStream<A>,
+    A: Actor,
+    Fn: FnMut(&S::Item, &mut A, &mut A::Context) -> Fut,
+    Fut: IntoActorFuture<A, Output = bool>,
+{
+    SkipWhile {
+        stream,
+        f,
+        pending_fut: None,
+        pending_item: None,
+        done_skipping: false,
+    }
+}
+
+impl<S, A, Fn, Fut> ActorStream<A> for SkipWhile<S, S::Item, Fn, Fut::Future>
+where
+    S: ActorStream<A>,
+    A: Actor,
+    Fn: FnMut(&S::Item, &mut A, &mut A::Context) -> Fut,
+    Fut: IntoActorFuture<A, Output = bool>,
+{
+    type Item = S::Item;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        act: &mut A,
+        ctx: &mut A::Context,
+        task: &mut task::Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        let mut this = self.project();
+
+        if *this.done_skipping {
+            return this.stream.poll_next(act, ctx, task);
+        }
+
+        Poll::Ready(loop {
+            if let Some(fut) = this.pending_fut.as_mut().as_pin_mut() {
+                let skipped = ready!(fut.poll(act, ctx, task));
+                let item = this.pending_item.take();
+                this.pending_fut.set(None);
+                if !skipped {
+                    *this.done_skipping = true;
+                    break item;
+                }
+            } else if let Some(item) = ready!(this.stream.as_mut().poll_next(act, ctx, task)) {
+                this.pending_fut
+                    .set(Some((this.f)(&item, act, ctx).into_future()));
+                *this.pending_item = Some(item);
+            } else {
+                break None;
+            }
+        })
+    }
+}
diff --git a/actix/src/fut/stream/take_while.rs b/actix/src/fut/stream/take_while.rs
new file mode 100644
index 000000000..10fa8682f
--- /dev/null
+++ b/actix/src/fut/stream/take_while.rs
@@ -0,0 +1,82 @@
+use std::pin::Pin;
+use std::task::{self, Poll};
+
+use futures_core::ready;
+use pin_project_lite::pin_project;
+
+use crate::actor::Actor;
+use crate::fut::{ActorFuture, ActorStream, IntoActorFuture};
+
+pin_project! {
+    /// Stream for the [`take_while`](super::ActorStreamExt::take_while) method.
+    #[must_use = "streams do nothing unless polled"]
+    #[derive(Debug)]
+    pub struct TakeWhile<S, I, Fut, Fn> {
+        #[pin]
+        stream: S,
+        f: Fn,
+        #[pin]
+        pending_fut: Option<Fut>,
+        pending_item: Option<I>,
+        done_taking: bool,
+    }
+}
+
+pub(super) fn new<S, A, Fut, Fn>(stream: S, f: Fn) -> TakeWhile<S, S::Item, Fut::Future, Fn>
+where
+    S: ActorStream<A>,
+    A: Actor,
+    Fn: FnMut(&S::Item, &mut A, &mut A::Context) -> Fut,
+    Fut: IntoActorFuture<A, Output = bool>,
+{
+    TakeWhile {
+        stream,
+        f,
+        pending_fut: None,
+        pending_item: None,
+        done_taking: false,
+    }
+}
+
+impl<S, A, Fut, Fn> ActorStream<A> for TakeWhile<S, S::Item, Fut::Future, Fn>
+where
+    S: ActorStream<A>,
+    A: Actor,
+    Fn: FnMut(&S::Item, &mut A, &mut A::Context) -> Fut,
+    Fut: IntoActorFuture<A, Output = bool>,
+{
+    type Item = S::Item;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        act: &mut A,
+        ctx: &mut A::Context,
+        task: &mut task::Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        if self.done_taking {
+            return Poll::Ready(None);
+        }
+
+        let mut this = self.project();
+
+        Poll::Ready(loop {
+            if let Some(fut) = this.pending_fut.as_mut().as_pin_mut() {
+                let take = ready!(fut.poll(act, ctx, task));
+                let item = this.pending_item.take();
+                this.pending_fut.set(None);
+                if take {
+                    break item;
+                } else {
+                    *this.done_taking = true;
+                    break None;
+                }
+            } else if let Some(item) = ready!(this.stream.as_mut().poll_next(act, ctx, task)) {
+                this.pending_fut
+                    .set(Some((this.f)(&item, act, ctx).into_future()));
+                *this.pending_item = Some(item);
+            } else {
+                break None;
+            }
+        })
+    }
+}
diff --git a/actix/src/fut/stream/then.rs b/actix/src/fut/stream/then.rs
new file mode 100644
index 000000000..7943fbd2a
--- /dev/null
+++ b/actix/src/fut/stream/then.rs
@@ -0,0 +1,67 @@
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use futures_core::ready;
+use pin_project_lite::pin_project;
+
+use crate::actor::Actor;
+use crate::fut::{ActorFuture, ActorStream, IntoActorFuture};
+
+pin_project! {
+    /// Stream for the [`then`](super::ActorStreamExt::then) method.
+    #[derive(Debug)]
+    #[must_use = "streams do nothing unless polled"]
+    pub struct Then<S, Fn, F> {
+        #[pin]
+        stream: S,
+        #[pin]
+        future: Option<F>,
+        f: Fn,
+    }
+}
+
+pub(super) fn new<S, A, Fn, F>(stream: S, f: Fn) -> Then<S, Fn, F::Future>
+where
+    S: ActorStream<A>,
+    A: Actor,
+    F: IntoActorFuture<A>,
+    Fn: FnMut(S::Item, &mut A, &mut A::Context) -> F,
+{
+    Then {
+        stream,
+        f,
+        future: None,
+    }
+}
+
+impl<S, A, Fn, F> ActorStream<A> for Then<S, Fn, F::Future>
+where
+    S: ActorStream<A>,
+    A: Actor,
+    F: IntoActorFuture<A>,
+    Fn: FnMut(S::Item, &mut A, &mut A::Context) -> F,
+{
+    type Item = F::Output;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        act: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        let mut this = self.project();
+
+        Poll::Ready(loop {
+            if let Some(fut) = this.future.as_mut().as_pin_mut() {
+                let item = ready!(fut.poll(act, ctx, task));
+                this.future.set(None);
+                break Some(item);
+            } else if let Some(item) = ready!(this.stream.as_mut().poll_next(act, ctx, task)) {
+                this.future
+                    .set(Some((this.f)(item, act, ctx).into_future()));
+            } else {
+                break None;
+            }
+        })
+    }
+}
diff --git a/actix/src/fut/stream/timeout.rs b/actix/src/fut/stream/timeout.rs
new file mode 100644
index 000000000..2963b165a
--- /dev/null
+++ b/actix/src/fut/stream/timeout.rs
@@ -0,0 +1,70 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::time::Duration;
+
+use pin_project_lite::pin_project;
+
+use crate::actor::Actor;
+use crate::clock::{sleep, Instant, Sleep};
+use crate::fut::ActorStream;
+
+pin_project! {
+    /// Stream for the [`timeout`](super::ActorStreamExt::timeout) method.
+    #[derive(Debug)]
+    #[must_use = "streams do nothing unless polled"]
+    pub struct Timeout<S> {
+        #[pin]
+        stream: S,
+        dur: Duration,
+        reset_timeout: bool,
+        #[pin]
+        timeout: Sleep,
+    }
+}
+
+impl<S> Timeout<S> {
+    pub(super) fn new(stream: S, timeout: Duration) -> Self {
+        Self {
+            stream,
+            dur: timeout,
+            reset_timeout: false,
+            timeout: sleep(timeout),
+        }
+    }
+}
+
+impl<S, A> ActorStream<A> for Timeout<S>
+where
+    S: ActorStream<A>,
+    A: Actor,
+{
+    type Item = Result<S::Item, ()>;
+
+    fn poll_next(
+        self: Pin<&mut Self>,
+        act: &mut A,
+        ctx: &mut A::Context,
+        task: &mut Context<'_>,
+    ) -> Poll<Option<Result<S::Item, ()>>> {
+        let mut this = self.project();
+
+        match this.stream.poll_next(act, ctx, task) {
+            Poll::Ready(Some(res)) => {
+                *this.reset_timeout = true;
+                Poll::Ready(Some(Ok(res)))
+            }
+            Poll::Ready(None) => Poll::Ready(None),
+            Poll::Pending => {
+                // only reset timeout when poll_next returns Ready and followed by Pending after.
+                if *this.reset_timeout {
+                    *this.reset_timeout = false;
+                    this.timeout.as_mut().reset(Instant::now() + *this.dur);
+                }
+
+                // check timeout
+                this.timeout.poll(task).map(|_| Some(Err(())))
+            }
+        }
+    }
+}
diff --git a/actix/src/fut/stream_finish.rs b/actix/src/fut/stream_finish.rs
deleted file mode 100644
index 37b74d965..000000000
--- a/actix/src/fut/stream_finish.rs
+++ /dev/null
@@ -1,47 +0,0 @@
-use std::pin::Pin;
-use std::task::{Context, Poll};
-
-use pin_project_lite::pin_project;
-
-use crate::actor::Actor;
-use crate::fut::{ActorFuture, ActorStream};
-
-pin_project! {
-    /// A combinator used to convert stream into a future, future resolves
-    /// when stream completes.
-    ///
-    /// This structure is produced by the `ActorStream::finish` method.
-    #[must_use = "streams do nothing unless polled"]
-    #[derive(Debug)]
-    pub struct StreamFinish<S> {
-        #[pin]
-        stream: S
-    }
-}
-
-pub fn new<S>(stream: S) -> StreamFinish<S> {
-    StreamFinish { stream }
-}
-
-impl<S, A> ActorFuture<A> for StreamFinish<S>
-where
-    S: ActorStream<A>,
-    A: Actor,
-{
-    type Output = ();
-
-    fn poll(
-        mut self: Pin<&mut Self>,
-        act: &mut A,
-        ctx: &mut A::Context,
-        task: &mut Context<'_>,
-    ) -> Poll<()> {
-        loop {
-            match self.as_mut().project().stream.poll_next(act, ctx, task) {
-                Poll::Pending => return Poll::Pending,
-                Poll::Ready(None) => return Poll::Ready(()),
-                Poll::Ready(Some(_)) => (),
-            };
-        }
-    }
-}
diff --git a/actix/src/fut/stream_fold.rs b/actix/src/fut/stream_fold.rs
deleted file mode 100644
index 13a7e4485..000000000
--- a/actix/src/fut/stream_fold.rs
+++ /dev/null
@@ -1,104 +0,0 @@
-use std::pin::Pin;
-use std::task::{Context, Poll};
-
-use pin_project_lite::pin_project;
-
-use crate::actor::Actor;
-use crate::fut::{ActorFuture, ActorStream, IntoActorFuture};
-
-pin_project! {
-    /// A future used to collect all the results of a stream into one generic type.
-    ///
-    /// This future is returned by the `ActorStream::fold` method.
-    #[derive(Debug)]
-    #[must_use = "streams do nothing unless polled"]
-    pub struct StreamFold<S, F, Fut, T> {
-        #[pin]
-        stream: S,
-        f: F,
-        #[pin]
-        state: State<T, Fut>,
-    }
-}
-
-pin_project! {
-    #[project = FoldStateProj]
-    #[derive(Debug)]
-    enum State<T, F> {
-        /// Placeholder state when doing work
-        Empty,
-
-        /// Ready to process the next stream item; current accumulator is the `T`
-        Ready {
-            res: Option<T>
-        },
-
-        /// Working on a future the process the previous stream item
-        Processing {
-            #[pin]
-            fut: F
-        },
-    }
-}
-
-pub fn new<S, A, F, Fut, T>(stream: S, f: F, t: T) -> StreamFold<S, F, Fut::Future, T>
-where
-    S: ActorStream<A>,
-    A: Actor,
-    F: FnMut(T, S::Item, &mut A, &mut A::Context) -> Fut,
-    Fut: IntoActorFuture<A, Output = T>,
-{
-    StreamFold {
-        stream,
-        f,
-        state: State::Ready { res: Some(t) },
-    }
-}
-
-impl<S, A, F, Fut, T> ActorFuture<A> for StreamFold<S, F, Fut::Future, T>
-where
-    S: ActorStream<A>,
-    A: Actor,
-    F: FnMut(T, S::Item, &mut A, &mut A::Context) -> Fut,
-    Fut: IntoActorFuture<A, Output = T>,
-    Fut::Future: ActorFuture<A>,
-{
-    type Output = T;
-
-    fn poll(
-        mut self: Pin<&mut Self>,
-        act: &mut A,
-        ctx: &mut A::Context,
-        task: &mut Context<'_>,
-    ) -> Poll<T> {
-        loop {
-            let this = self.as_mut().project();
-            match this.state.project() {
-                FoldStateProj::Ready { res } => match this.stream.poll_next(act, ctx, task) {
-                    Poll::Ready(Some(e)) => {
-                        let future = (this.f)(res.take().unwrap(), e, act, ctx);
-                        let fut = future.into_future();
-                        self.as_mut().project().state.set(State::Processing { fut });
-                    }
-                    Poll::Ready(None) => {
-                        return {
-                            let res = res.take().unwrap();
-                            self.project().state.set(State::Empty);
-                            Poll::Ready(res)
-                        }
-                    }
-                    Poll::Pending => return Poll::Pending,
-                },
-                FoldStateProj::Processing { fut } => match fut.poll(act, ctx, task) {
-                    Poll::Ready(state) => self
-                        .as_mut()
-                        .project()
-                        .state
-                        .set(State::Ready { res: Some(state) }),
-                    Poll::Pending => return Poll::Pending,
-                },
-                FoldStateProj::Empty => panic!("cannot poll Fold twice"),
-            }
-        }
-    }
-}
diff --git a/actix/src/fut/stream_map.rs b/actix/src/fut/stream_map.rs
deleted file mode 100644
index f6f4d010c..000000000
--- a/actix/src/fut/stream_map.rs
+++ /dev/null
@@ -1,58 +0,0 @@
-use std::pin::Pin;
-use std::task::{Context, Poll};
-
-use pin_project_lite::pin_project;
-
-use crate::actor::Actor;
-use crate::fut::ActorStream;
-
-pin_project! {
-    /// A stream combinator which will change the type of a stream from one
-    /// type to another.
-    ///
-    /// This is produced by the `ActorStream::map` method.
-    #[derive(Debug)]
-    #[must_use = "streams do nothing unless polled"]
-    pub struct StreamMap<S, F> {
-        #[pin]
-        stream: S,
-        f: F,
-    }
-}
-
-pub fn new<S, A, F, U>(stream: S, f: F) -> StreamMap<S, F>
-where
-    F: FnMut(S::Item, &mut A, &mut A::Context) -> U,
-    S: ActorStream<A>,
-    A: Actor,
-{
-    StreamMap { stream, f }
-}
-
-impl<S, A, F, U> ActorStream<A> for StreamMap<S, F>
-where
-    S: ActorStream<A>,
-    A: Actor,
-    F: FnMut(S::Item, &mut A, &mut A::Context) -> U,
-{
-    type Item = U;
-
-    fn poll_next(
-        self: Pin<&mut Self>,
-        act: &mut A,
-        ctx: &mut A::Context,
-        task: &mut Context<'_>,
-    ) -> Poll<Option<U>> {
-        let this = self.project();
-        match this.stream.poll_next(act, ctx, task) {
-            Poll::Pending => Poll::Pending,
-            Poll::Ready(option) => {
-                if let Some(item) = option {
-                    Poll::Ready(Some((this.f)(item, act, ctx)))
-                } else {
-                    Poll::Ready(None)
-                }
-            }
-        }
-    }
-}
diff --git a/actix/src/fut/stream_then.rs b/actix/src/fut/stream_then.rs
deleted file mode 100644
index 9bdec6b4d..000000000
--- a/actix/src/fut/stream_then.rs
+++ /dev/null
@@ -1,78 +0,0 @@
-use std::pin::Pin;
-use std::task::{Context, Poll};
-
-use pin_project_lite::pin_project;
-
-use crate::actor::Actor;
-use crate::fut::{ActorFuture, ActorStream, IntoActorFuture};
-
-pin_project! {
-    /// A stream combinator which chains a computation onto each item produced by a
-    /// stream.
-    ///
-    /// This structure is produced by the `ActorStream::then` method.
-    #[derive(Debug)]
-    #[must_use = "streams do nothing unless polled"]
-    pub struct StreamThen<S, Fn, F> {
-        #[pin]
-        stream: S,
-        #[pin]
-        future: Option<F>,
-        f: Fn,
-    }
-}
-
-pub fn new<S, A, Fn, F>(stream: S, f: Fn) -> StreamThen<S, Fn, F::Future>
-where
-    S: ActorStream<A>,
-    A: Actor,
-    F: IntoActorFuture<A>,
-    Fn: FnMut(S::Item, &mut A, &mut A::Context) -> F,
-{
-    StreamThen {
-        stream,
-        f,
-        future: None,
-    }
-}
-
-impl<S, A, Fn, F> ActorStream<A> for StreamThen<S, Fn, F::Future>
-where
-    S: ActorStream<A>,
-    A: Actor,
-    F: IntoActorFuture<A>,
-    Fn: FnMut(S::Item, &mut A, &mut A::Context) -> F,
-{
-    type Item = F::Output;
-
-    fn poll_next(
-        mut self: Pin<&mut Self>,
-        act: &mut A,
-        ctx: &mut A::Context,
-        task: &mut Context<'_>,
-    ) -> Poll<Option<Self::Item>> {
-        loop {
-            let this = self.as_mut().project();
-            match this.future.as_pin_mut() {
-                None => {
-                    let item = match this.stream.poll_next(act, ctx, task) {
-                        Poll::Pending => return Poll::Pending,
-                        Poll::Ready(None) => return Poll::Ready(None),
-                        Poll::Ready(Some(e)) => e,
-                    };
-                    let fut = (this.f)(item, act, ctx).into_future();
-                    self.as_mut().project().future.set(Some(fut));
-                }
-                Some(fut) => {
-                    return match fut.poll(act, ctx, task) {
-                        Poll::Ready(e) => {
-                            self.project().future.set(None);
-                            Poll::Ready(Some(e))
-                        }
-                        Poll::Pending => Poll::Pending,
-                    }
-                }
-            }
-        }
-    }
-}
diff --git a/actix/src/fut/stream_timeout.rs b/actix/src/fut/stream_timeout.rs
deleted file mode 100644
index a3ec7ac1b..000000000
--- a/actix/src/fut/stream_timeout.rs
+++ /dev/null
@@ -1,83 +0,0 @@
-use std::future::Future;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use std::time::Duration;
-
-use pin_project_lite::pin_project;
-
-use crate::actor::Actor;
-use crate::clock::{self, Sleep};
-use crate::fut::ActorStream;
-
-pin_project! {
-    /// Future for the `timeout` combinator, interrupts computations if it takes
-    /// more than `timeout`.
-    ///
-    /// This is created by the `ActorFuture::timeout()` method.
-    #[derive(Debug)]
-    #[must_use = "streams do nothing unless polled"]
-    pub struct StreamTimeout<S> {
-        #[pin]
-        stream: S,
-        dur: Duration,
-        #[pin]
-        timeout: Option<Sleep>,
-    }
-}
-
-pub fn new<S, A>(stream: S, timeout: Duration) -> StreamTimeout<S>
-where
-    S: ActorStream<A>,
-    A: Actor,
-{
-    StreamTimeout {
-        stream,
-        dur: timeout,
-        timeout: None,
-    }
-}
-
-impl<S, A> ActorStream<A> for StreamTimeout<S>
-where
-    S: ActorStream<A>,
-    A: Actor,
-{
-    type Item = Result<S::Item, ()>;
-
-    fn poll_next(
-        self: Pin<&mut Self>,
-        act: &mut A,
-        ctx: &mut A::Context,
-        task: &mut Context<'_>,
-    ) -> Poll<Option<Result<S::Item, ()>>> {
-        let mut this = self.project();
-
-        match this.stream.poll_next(act, ctx, task) {
-            Poll::Ready(Some(res)) => {
-                this.timeout.set(None);
-                return Poll::Ready(Some(Ok(res)));
-            }
-            Poll::Ready(None) => return Poll::Ready(None),
-            Poll::Pending => (),
-        }
-
-        if this.timeout.is_none() {
-            this.timeout.set(Some(clock::sleep(*this.dur)));
-        }
-
-        // check timeout
-        if this
-            .timeout
-            .as_mut()
-            .as_pin_mut()
-            .unwrap()
-            .poll(task)
-            .is_pending()
-        {
-            return Poll::Pending;
-        }
-        this.timeout.set(None);
-
-        Poll::Ready(Some(Err(())))
-    }
-}
diff --git a/actix/src/fut/then.rs b/actix/src/fut/then.rs
deleted file mode 100644
index 0193f9cd2..000000000
--- a/actix/src/fut/then.rs
+++ /dev/null
@@ -1,60 +0,0 @@
-use std::pin::Pin;
-use std::task::{self, Poll};
-
-use pin_project_lite::pin_project;
-
-use crate::actor::Actor;
-use crate::fut::chain::Chain;
-use crate::fut::{ActorFuture, IntoActorFuture};
-
-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.
-    #[derive(Debug)]
-    #[must_use = "futures do nothing unless polled"]
-    pub struct Then<A, B, Fn, Act>
-    where
-        A: ActorFuture<Act>,
-        B: IntoActorFuture<Act>,
-        Act: Actor,
-    {
-        #[pin]
-        state: Chain<A, B::Future, Fn, Act>,
-    }
-}
-
-pub fn new<A, B, Fn, Act>(future: A, f: Fn) -> Then<A, B, Fn, Act>
-where
-    A: ActorFuture<Act>,
-    B: IntoActorFuture<Act>,
-    Act: Actor,
-{
-    Then {
-        state: Chain::new(future, f),
-    }
-}
-
-impl<A, B, Fn, Act> ActorFuture<Act> for Then<A, B, Fn, Act>
-where
-    A: ActorFuture<Act>,
-    B: IntoActorFuture<Act>,
-    Fn: FnOnce(A::Output, &mut Act, &mut Act::Context) -> B,
-    Act: Actor,
-{
-    type Output = B::Output;
-
-    fn poll(
-        self: Pin<&mut Self>,
-        act: &mut Act,
-        ctx: &mut Act::Context,
-        task: &mut task::Context<'_>,
-    ) -> Poll<B::Output> {
-        self.project()
-            .state
-            .poll(act, ctx, task, |item, f, act, ctx| {
-                f(item, act, ctx).into_future()
-            })
-    }
-}
diff --git a/actix/src/handler.rs b/actix/src/handler.rs
index ef020adfd..06955b751 100644
--- a/actix/src/handler.rs
+++ b/actix/src/handler.rs
@@ -4,7 +4,7 @@ pub use tokio::sync::oneshot::Sender as OneshotSender;
 
 use crate::actor::{Actor, AsyncContext};
 use crate::address::Addr;
-use crate::fut::ActorFuture;
+use crate::fut::{ActorFuture, ActorFutureExt, LocalBoxActorFuture};
 
 /// Describes how to handle messages of a specific type.
 ///
@@ -198,7 +198,7 @@ where
 ///     }
 /// }
 /// ```
-pub type ResponseActFuture<A, I> = Pin<Box<dyn ActorFuture<A, Output = I>>>;
+pub type ResponseActFuture<A, I> = LocalBoxActorFuture<A, I>;
 
 /// A specialized future for asynchronous message handling.
 ///
diff --git a/actix/src/lib.rs b/actix/src/lib.rs
index af4e1079d..706e60161 100644
--- a/actix/src/lib.rs
+++ b/actix/src/lib.rs
@@ -60,7 +60,9 @@ pub use crate::actor::{
 };
 pub use crate::address::{Addr, MailboxError, Recipient, WeakAddr, WeakRecipient};
 pub use crate::context::Context;
-pub use crate::fut::{ActorFuture, ActorStream, FinishStream, WrapFuture, WrapStream};
+pub use crate::fut::{
+    ActorFuture, ActorFutureExt, ActorStream, ActorStreamExt, WrapFuture, WrapStream,
+};
 pub use crate::handler::{
     ActorResponse, AtomicResponse, Handler, Message, MessageResult, Response,
     ResponseActFuture, ResponseFuture,
@@ -97,7 +99,9 @@ pub mod prelude {
         Addr, MailboxError, Recipient, RecipientRequest, Request, SendError,
     };
     pub use crate::context::{Context, ContextFutureSpawner};
-    pub use crate::fut::{ActorFuture, ActorStream, WrapFuture, WrapStream};
+    pub use crate::fut::{
+        ActorFuture, ActorFutureExt, ActorStream, ActorStreamExt, WrapFuture, WrapStream,
+    };
     pub use crate::handler::{
         ActorResponse, AtomicResponse, Handler, Message, MessageResult, Response,
         ResponseActFuture, ResponseFuture,


Test patch:
diff --git a/actix/tests/test_fut.rs b/actix/tests/test_fut.rs
index 364c0d65a..1bd9b7e59 100644
--- a/actix/tests/test_fut.rs
+++ b/actix/tests/test_fut.rs
@@ -45,6 +45,7 @@ fn test_fut_timeout() {
 }
 
 struct MyStreamActor {
+    counter: usize,
     timeout: Arc<AtomicBool>,
 }
 
@@ -60,13 +61,11 @@ impl Actor for MyStreamActor {
             .timeout(Duration::new(0, 1000))
             .then(|res, act, _| {
                 // Additional waiting time to test `then` call as well
-                Box::pin(
-                    async move {
-                        sleep(Duration::from_millis(500)).await;
-                        res
-                    }
-                    .into_actor(act),
-                )
+                async move {
+                    sleep(Duration::from_millis(500)).await;
+                    res
+                }
+                .into_actor(act)
             })
             .map(|e, act, _| {
                 if let Err(()) = e {
@@ -79,6 +78,96 @@ impl Actor for MyStreamActor {
     }
 }
 
+struct TakeWhileMsg(usize);
+
+impl Message for TakeWhileMsg {
+    type Result = ();
+}
+
+impl Handler<TakeWhileMsg> for MyStreamActor {
+    type Result = ResponseActFuture<Self, ()>;
+
+    fn handle(&mut self, msg: TakeWhileMsg, _: &mut Context<Self>) -> Self::Result {
+        let num = msg.0;
+        Box::pin(
+            futures_util::stream::repeat(num)
+                .into_actor(self)
+                .take_while(move |n, act, ctx| {
+                    ctx.spawn(
+                        async {
+                            actix_rt::task::yield_now().await;
+                        }
+                        .into_actor(act),
+                    );
+                    assert_eq!(*n, num);
+                    assert!(act.counter < 10);
+                    act.counter += 1;
+                    futures_util::future::ready(act.counter < 10)
+                })
+                .finish(),
+        )
+    }
+}
+
+struct SkipWhileMsg(usize);
+
+impl Message for SkipWhileMsg {
+    type Result = ();
+}
+
+impl Handler<SkipWhileMsg> for MyStreamActor {
+    type Result = ResponseActFuture<Self, ()>;
+
+    fn handle(&mut self, msg: SkipWhileMsg, _: &mut Context<Self>) -> Self::Result {
+        let num = msg.0;
+        Box::pin(
+            futures_util::stream::repeat(num)
+                .into_actor(self)
+                .take_while(|_, act, _| {
+                    let cond = act.counter < 10;
+                    act.counter += 1;
+                    futures_util::future::ready(cond)
+                })
+                .skip_while(move |n, act, ctx| {
+                    let fut = async {
+                        actix_rt::task::yield_now().await;
+                    }
+                    .into_actor(act);
+                    ctx.spawn(fut);
+                    assert_eq!(*n, num);
+                    act.counter -= 1;
+                    futures_util::future::ready(act.counter > 0)
+                })
+                .finish(),
+        )
+    }
+}
+
+struct CollectMsg(usize);
+
+impl Message for CollectMsg {
+    type Result = Vec<usize>;
+}
+
+impl Handler<CollectMsg> for MyStreamActor {
+    type Result = ResponseActFuture<Self, Vec<usize>>;
+
+    fn handle(&mut self, msg: CollectMsg, _: &mut Context<Self>) -> Self::Result {
+        let num = msg.0;
+        Box::pin(
+            futures_util::stream::repeat(num)
+                .into_actor(self)
+                .take_while(|_, act, _| {
+                    let cond = act.counter < 5;
+                    act.counter += 1;
+                    futures_util::future::ready(cond)
+                })
+                .map(|_, act, _| act.counter)
+                .collect(),
+        )
+    }
+}
+
 #[test]
 fn test_stream_timeout() {
     let timeout = Arc::new(AtomicBool::new(false));
@@ -86,9 +175,51 @@ fn test_stream_timeout() {
 
     let sys = System::new();
     sys.block_on(async {
-        let _addr = MyStreamActor { timeout: timeout2 }.start();
+        let _addr = MyStreamActor {
+            timeout: timeout2,
+            counter: 0,
+        }
+        .start();
     });
     sys.run().unwrap();
 
     assert!(timeout.load(Ordering::Relaxed), "Not timeout");
 }
+
+#[test]
+fn test_stream_take_while() {
+    System::new().block_on(async {
+        let addr = MyStreamActor {
+            timeout: Arc::new(AtomicBool::new(false)),
+            counter: 0,
+        }
+        .start();
+        addr.send(TakeWhileMsg(5)).await.unwrap();
+    })
+}
+
+#[test]
+fn test_stream_skip_while() {
+    System::new().block_on(async {
+        let addr = MyStreamActor {
+            timeout: Arc::new(AtomicBool::new(false)),
+            counter: 0,
+        }
+        .start();
+        addr.send(SkipWhileMsg(5)).await.unwrap();
+    })
+}
+
+#[test]
+fn test_stream_collect() {
+    System::new().block_on(async {
+        let addr = MyStreamActor {
+            timeout: Arc::new(AtomicBool::new(false)),
+            counter: 0,
+        }
+        .start();
+        let res = addr.send(CollectMsg(3)).await.unwrap();
+
+        assert_eq!(res, vec![1, 2, 3, 4, 5]);
+    })
+}