Repository: actix/actix (language: rust)

Issue:
Update `tokio` to 1.0
New version of tokio is released
tgbot vesrion of tokio is 0.2 [Cargo.toml](https://github.com/actix/actix/blob/master/Cargo.toml)

Better upgrade, to avoid downloading two versions of tokio in projects


Discussion / hints:
I'm happy to do it but let me take a look at the upstream dependencies.
The main blocker seems to be actix-server which is still using mio 0.6.
The latest version has some serious breaking change with some features gone so some big changes and extra deps could be needed.

Fix patch:
diff --git a/.github/workflows/linux.yml b/.github/workflows/linux.yml
index 16ccf02b5..2b55fb954 100644
--- a/.github/workflows/linux.yml
+++ b/.github/workflows/linux.yml
@@ -14,7 +14,7 @@ jobs:
       fail-fast: false
       matrix:
         version:
-          - 1.40.0 # MSRV
+          - 1.46.0 # MSRV
           - stable
           - nightly
 
diff --git a/.gitignore b/.gitignore
index c19752815..11cec9ac3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -10,3 +10,4 @@ __pycache__
 *~
 target/
 examples/chat/target/
+.idea
\ No newline at end of file
diff --git a/CHANGES.md b/CHANGES.md
index c13505777..b76b4b2d6 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -17,6 +17,13 @@
   longer bound to Actor. [#445]
 * Re-export `actix_rt::main` macro as `actix::main` [#448]
 * Bump `crossbeam-channel` to `0.5`
+* Update `bytes` to `1`.
+* Update `tokio` to `1`.
+* Update `tokio-util` tp `0.6`.
+* Rename `actix::clock::{delay_for, delay_until, Delay}` to `{sleep, sleep_until, Sleep}`
+* Remove all `Unpin` requirement from `ActorStream`.
+* Remove `actix::clock::Duration` type. It's a re-export of `std::time::Duration` and use 
+  the Duration type in std lib is suggested.
 * Update examples and tests according to the change of `actix-rt`. 
   `Arbiter::spawn` and `actix_rt::spawn` would panic outside the context of `actix::System`. They must be call
   inside `System::run`,`SystemRunner::run` or `SystemRunner::block_on`.
diff --git a/Cargo.toml b/Cargo.toml
index 75f7e8009..8103e48ea 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -33,30 +33,31 @@ derive = ["actix_derive"]
 mailbox_assert = []
 
 [dependencies]
-actix-rt = "1.1.1"
+actix-rt = "2.0.0-beta.1"
 actix_derive = { version = "0.5", optional = true }
 actix-macros = { version = "0.1.3", features = ["actix-reexport"] }
 
-bytes = "0.5.3"
+bytes = "1"
 crossbeam-channel = "0.5"
 derive_more = "0.99.2"
-futures-channel = { version = "0.3.1", default-features = false }
-futures-util = { version = "0.3.1", default-features = false, features = ["sink"] }
+futures-channel = { version = "0.3.7", default-features = false }
+futures-util = { version = "0.3.7", default-features = false, features = ["sink"] }
 log = "0.4"
 pin-project = "1.0.0"
 once_cell = "1.4"
 bitflags = "1.2"
 smallvec = "1.4"
 parking_lot = "0.11"
-tokio = { version = "0.2.6", default-features = false, features = ["rt-core", "rt-util", "io-driver", "io-util", "tcp", "uds", "udp", "time", "signal", "sync"] }
-tokio-util = { version = "0.3", features = ["full"] }
+tokio = { version = "1", features = ["io-util"] }
+tokio-util = { version = "0.6", features = ["full"] }
 
 # dns resolver
-trust-dns-proto = { version = "0.19", optional = true, default-features = false, features = ["tokio-runtime"] }
-trust-dns-resolver = { version = "0.19", optional = true, default-features = false, features = ["tokio-runtime", "system-config"] }
+trust-dns-proto = { version = "0.20.0" , optional = true, default-features = false, features = ["tokio-runtime"] }
+trust-dns-resolver = { version = "0.20.0" , optional = true, default-features = false, features = ["tokio-runtime", "system-config"] }
 
 [dev-dependencies]
 doc-comment = "0.3"
+futures-channel = { version = "0.3.7", default-features = false, features = ["sink"] }
 
 [profile.release]
 lto = true
diff --git a/examples/chat/Cargo.toml b/examples/chat/Cargo.toml
index cb36d4a6c..ab5622f30 100644
--- a/examples/chat/Cargo.toml
+++ b/examples/chat/Cargo.toml
@@ -14,17 +14,16 @@ name = "client"
 path = "src/client.rs"
 
 [dependencies]
-bytes = "0.5"
+actix = { path = "../../" }
+actix_derive = "0.5"
+actix-rt = "2.0.0-beta.1"
+
+async-stream = "0.3"
+bytes = "1"
 byteorder = "1.3"
-futures-util = "0.3"
+futures-util = "0.3.7"
 rand = "0.7"
-
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-
-tokio = { version = "0.2", default-features=false, features=["tcp", "time", "signal", "io-util"] }
-tokio-util = { version = "0.3", features = ["codec"] }
-
-actix = { path = "../../" }
-actix_derive = "0.5"
-actix-rt = "1.0"
+tokio = "1"
+tokio-util = { version = "0.6", features = ["codec"] }
diff --git a/examples/chat/src/main.rs b/examples/chat/src/main.rs
index 3b7a1f1b4..5e117e948 100644
--- a/examples/chat/src/main.rs
+++ b/examples/chat/src/main.rs
@@ -3,7 +3,6 @@ use std::net;
 use std::str::FromStr;
 
 use actix::prelude::*;
-use futures_util::stream::StreamExt;
 use tokio::net::{TcpListener, TcpStream};
 use tokio_util::codec::FramedRead;
 
@@ -56,7 +55,15 @@ async fn main() {
 
     // Create server listener
     let addr = net::SocketAddr::from_str("127.0.0.1:12345").unwrap();
-    let listener = Box::new(TcpListener::bind(&addr).await.unwrap());
+    let listener = TcpListener::bind(&addr).await.unwrap();
+
+    // create a stream with the help of stream macro.
+    let stream = async_stream::stream! {
+        // accept incoming tcp stream and remote addr and return them as a stream
+        while let Ok((st, addr)) = listener.accept().await {
+            yield TcpConnect(st, addr);
+        }
+    };
 
     // Our chat server `Server` is an actor, first we need to start it
     // and then add stream on incoming tcp connections to it.
@@ -64,11 +71,7 @@ async fn main() {
     // items So to be able to handle this events `Server` actor has to implement
     // stream handler `StreamHandler<(TcpStream, net::SocketAddr), io::Error>`
     Server::create(move |ctx| {
-        ctx.add_message_stream(Box::leak(listener).incoming().map(|st| {
-            let st = st.unwrap();
-            let addr = st.peer_addr().unwrap();
-            TcpConnect(st, addr)
-        }));
+        ctx.add_message_stream(stream);
         Server { chat: server }
     });
 
diff --git a/examples/chat/src/session.rs b/examples/chat/src/session.rs
index 9268813ba..b93d0a5df 100644
--- a/examples/chat/src/session.rs
+++ b/examples/chat/src/session.rs
@@ -1,8 +1,9 @@
 //! `ClientSession` is an actor, it manages peer TCP connection and
 //! proxies commands from peer to `ChatServer`.
 use std::io;
+use std::time::Duration;
 
-use actix::clock::{Duration, Instant};
+use actix::clock::Instant;
 use actix::prelude::*;
 
 use tokio::io::WriteHalf;
diff --git a/src/actors/mod.rs b/src/actors/mod.rs
index 13bd80468..1a734ec7a 100644
--- a/src/actors/mod.rs
+++ b/src/actors/mod.rs
@@ -3,4 +3,8 @@
 pub mod mocker;
 
 #[cfg(feature = "resolver")]
+#[deprecated(
+    since = "0.20.0",
+    note = "resolver actor is deprecated and possibly removed in future release."
+)]
 pub mod resolver;
diff --git a/src/actors/resolver.rs b/src/actors/resolver.rs
index 6a0209af0..47976ba6e 100644
--- a/src/actors/resolver.rs
+++ b/src/actors/resolver.rs
@@ -1,8 +1,8 @@
 //! DNS resolver and connector utility actor
 //!
 //! ## Example
-//!
 //! ```rust
+//! #[allow(deprecated)]
 //! use actix::actors::resolver;
 //! use actix::prelude::*;
 //!
@@ -35,13 +35,12 @@ use std::task::{self, Poll};
 use std::time::Duration;
 
 use derive_more::Display;
-use log::warn;
 use tokio::net::TcpStream;
 use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
 use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
 use trust_dns_resolver::{error::ResolveError, lookup_ip::LookupIp};
 
-use crate::clock::Delay;
+use crate::clock::Sleep;
 use crate::fut::ActorFuture;
 use crate::fut::Either;
 use crate::prelude::*;
@@ -161,46 +160,22 @@ impl Actor for Resolver {
     type Context = Context<Self>;
     fn started(&mut self, ctx: &mut Self::Context) {
         let cfg = self.cfg.take();
-        ctx.wait(
-            async move { cfg }
-                .into_actor(self)
-                .then(
-                    |cfg, this, _| -> Pin<Box<dyn ActorFuture<Actor = Self, Output = _>>> {
-                        if let Some(cfg) = cfg {
-                            return Box::pin(
-                                AsyncResolver::tokio(cfg.0, cfg.1).into_actor(this),
-                            );
-                        }
-                        Box::pin(
-                            async {
-                                match AsyncResolver::from_system_conf(
-                                    tokio::runtime::Handle::current(),
-                                )
-                                .await
-                                {
-                                    Ok(resolver) => Ok(resolver),
-                                    Err(err) => {
-                                        warn!(
-                                            "Can not create system dns resolver: {}",
-                                            err
-                                        );
-                                        AsyncResolver::tokio(
-                                            ResolverConfig::default(),
-                                            ResolverOpts::default(),
-                                        )
-                                        .await
-                                    }
-                                }
-                            }
-                            .into_actor(this),
+        ctx.wait(async move { cfg }.into_actor(self).map(|cfg, this, _| {
+            let resolver = match cfg {
+                Some((conf, opt)) => AsyncResolver::tokio(conf, opt),
+                None => match AsyncResolver::tokio_from_system_conf() {
+                    Ok(resolver) => Ok(resolver),
+                    Err(err) => {
+                        log::warn!("Can not create system dns resolver: {}", err);
+                        AsyncResolver::tokio(
+                            ResolverConfig::default(),
+                            ResolverOpts::default(),
                         )
-                    },
-                )
-                .map(|resolver_res, this, _| {
-                    // Keep the resolver itself.
-                    this.resolver = Some(resolver_res.unwrap());
-                }),
-        );
+                    }
+                },
+            };
+            this.resolver = Some(resolver.unwrap())
+        }));
     }
 }
 
@@ -392,9 +367,11 @@ impl ActorFuture for ResolveFut {
 
 /// A TCP stream connector.
 #[allow(clippy::type_complexity)]
+#[pin_project::pin_project]
 pub struct TcpConnector {
     addrs: VecDeque<SocketAddr>,
-    timeout: Delay,
+    #[pin]
+    timeout: Sleep,
     stream: Option<Pin<Box<dyn Future<Output = Result<TcpStream, io::Error>>>>>,
 }
 
@@ -407,7 +384,7 @@ impl TcpConnector {
         TcpConnector {
             addrs,
             stream: None,
-            timeout: tokio::time::delay_for(timeout),
+            timeout: tokio::time::sleep(timeout),
         }
     }
 }
@@ -422,10 +399,10 @@ impl ActorFuture for TcpConnector {
         _: &mut Context<Resolver>,
         cx: &mut task::Context<'_>,
     ) -> Poll<Self::Output> {
-        let mut this = self.get_mut();
+        let this = self.project();
 
         // timeout
-        if let Poll::Ready(_) = Pin::new(&mut this.timeout).poll(cx) {
+        if this.timeout.poll(cx).is_ready() {
             return Poll::Ready(Err(ResolverError::Timeout));
         }
 
@@ -444,7 +421,7 @@ impl ActorFuture for TcpConnector {
             }
             // try to connect
             let addr = this.addrs.pop_front().unwrap();
-            this.stream = Some(Box::pin(TcpStream::connect(addr)));
+            *this.stream = Some(Box::pin(TcpStream::connect(addr)));
         }
     }
 }
diff --git a/src/address/message.rs b/src/address/message.rs
index 4a2b4304d..5f898074a 100644
--- a/src/address/message.rs
+++ b/src/address/message.rs
@@ -7,7 +7,7 @@ use futures_channel::oneshot;
 
 use pin_project::pin_project;
 
-use crate::clock::Delay;
+use crate::clock::Sleep;
 use crate::handler::{Handler, Message};
 
 use super::channel::{AddressSender, Sender};
@@ -25,7 +25,8 @@ where
 {
     rx: Option<oneshot::Receiver<M::Result>>,
     info: Option<(AddressSender<A>, M)>,
-    timeout: Option<Delay>,
+    #[pin]
+    timeout: Option<Sleep>,
 }
 
 impl<A, M> Request<A, M>
@@ -52,16 +53,16 @@ where
 
     /// Set message delivery timeout
     pub fn timeout(mut self, dur: Duration) -> Self {
-        self.timeout = Some(tokio::time::delay_for(dur));
+        self.timeout = Some(tokio::time::sleep(dur));
         self
     }
 
     fn poll_timeout(
-        &mut self,
+        self: Pin<&mut Self>,
         cx: &mut task::Context<'_>,
     ) -> Poll<Result<M::Result, MailboxError>> {
-        if let Some(ref mut timeout) = self.timeout {
-            match Pin::new(timeout).poll(cx) {
+        if let Some(timeout) = self.project().timeout.as_pin_mut() {
+            match timeout.poll(cx) {
                 Poll::Ready(()) => Poll::Ready(Err(MailboxError::Timeout)),
                 Poll::Pending => Poll::Pending,
             }
@@ -118,7 +119,8 @@ where
 {
     rx: Option<oneshot::Receiver<M::Result>>,
     info: Option<(Box<dyn Sender<M>>, M)>,
-    timeout: Option<Delay>,
+    #[pin]
+    timeout: Option<Sleep>,
 }
 
 impl<M> RecipientRequest<M>
@@ -139,16 +141,16 @@ where
 
     /// Set message delivery timeout
     pub fn timeout(mut self, dur: Duration) -> Self {
-        self.timeout = Some(tokio::time::delay_for(dur));
+        self.timeout = Some(tokio::time::sleep(dur));
         self
     }
 
     fn poll_timeout(
-        &mut self,
+        self: Pin<&mut Self>,
         cx: &mut task::Context<'_>,
     ) -> Poll<Result<M::Result, MailboxError>> {
-        if let Some(ref mut timeout) = self.timeout {
-            match Pin::new(timeout).poll(cx) {
+        if let Some(timeout) = self.project().timeout.as_pin_mut() {
+            match timeout.poll(cx) {
                 Poll::Ready(()) => Poll::Ready(Err(MailboxError::Timeout)),
                 Poll::Pending => Poll::Pending,
             }
diff --git a/src/clock.rs b/src/clock.rs
index 0cf5ffe72..9b48afbed 100644
--- a/src/clock.rs
+++ b/src/clock.rs
@@ -1,13 +1,9 @@
-//! A configurable source of time.
+//! Utilities for tracking time.
 //!
-//! This module provides an API to get the current instant in such a way that
-//! the source of time may be configured. This allows mocking out the source of
-//! time in tests.
+//! This module provides a number of types for executing code after a set period of time.
 //!
-//! See [Module `tokio_timer::clock`] for full documentation.
+//! See [Module `tokio::time`] for full documentation.
 //!
-//! [Module `tokio_timer::clock`]: https://docs.rs/tokio-timer/latest/tokio_timer/clock/index.html
+//! [Module `tokio::time`]: https://docs.rs/tokio/1.0.1/tokio/time/index.html
 
-pub use tokio::time::{
-    delay_for, delay_until, interval_at, Delay, Duration, Instant, Interval,
-};
+pub use actix_rt::time::*;
diff --git a/src/contextitems.rs b/src/contextitems.rs
index 4ce5d48c2..fe2389705 100644
--- a/src/contextitems.rs
+++ b/src/contextitems.rs
@@ -8,7 +8,7 @@ use futures_util::{ready, stream::Stream};
 use pin_project::pin_project;
 
 use crate::actor::{Actor, ActorContext, AsyncContext};
-use crate::clock::Delay;
+use crate::clock::Sleep;
 use crate::fut::ActorFuture;
 use crate::handler::{Handler, Message, MessageResponse};
 
@@ -48,22 +48,18 @@ where
     }
 }
 
+#[pin_project::pin_project]
 pub(crate) struct ActorDelayedMessageItem<A, M>
 where
     A: Actor,
     M: Message,
 {
     msg: Option<M>,
-    timeout: Delay,
+    #[pin]
+    timeout: Sleep,
     act: PhantomData<A>,
     m: PhantomData<M>,
 }
-impl<A, M> Unpin for ActorDelayedMessageItem<A, M>
-where
-    A: Actor,
-    M: Message,
-{
-}
 
 impl<A, M> ActorDelayedMessageItem<A, M>
 where
@@ -73,7 +69,7 @@ where
     pub fn new(msg: M, timeout: Duration) -> Self {
         Self {
             msg: Some(msg),
-            timeout: tokio::time::delay_for(timeout),
+            timeout: tokio::time::sleep(timeout),
             act: PhantomData,
             m: PhantomData,
         }
@@ -95,8 +91,8 @@ where
         ctx: &mut A::Context,
         task: &mut task::Context<'_>,
     ) -> Poll<Self::Output> {
-        let this = self.get_mut();
-        ready!(Pin::new(&mut this.timeout).poll(task));
+        let this = self.project();
+        ready!(this.timeout.poll(task));
         let fut = A::handle(act, this.msg.take().unwrap(), ctx);
         fut.handle::<()>(ctx, None);
         Poll::Ready(())
diff --git a/src/fut/mod.rs b/src/fut/mod.rs
index 79a55b16c..a4c189da7 100644
--- a/src/fut/mod.rs
+++ b/src/fut/mod.rs
@@ -224,7 +224,7 @@ pub trait ActorStream {
             &mut <Self::Actor as Actor>::Context,
         ) -> U,
         U: IntoActorFuture<Actor = Self::Actor>,
-        Self: Unpin + Sized,
+        Self: Sized,
     {
         stream_then::new(self, f)
     }
@@ -250,7 +250,7 @@ pub trait ActorStream {
     /// `err` value get returned as a timeout error.
     fn timeout(self, timeout: Duration) -> StreamTimeout<Self>
     where
-        Self: Sized + Unpin,
+        Self: Sized,
     {
         stream_timeout::new(self, timeout)
     }
@@ -258,7 +258,7 @@ pub trait ActorStream {
     /// Converts a stream to a future that resolves when stream finishes.
     fn finish(self) -> StreamFinish<Self>
     where
-        Self: Sized + Unpin,
+        Self: Sized,
     {
         stream_finish::new(self)
     }
@@ -416,7 +416,7 @@ where
     fn into_actor(self, a: &A) -> Self::Stream;
 }
 
-impl<S: Stream + Unpin, A: Actor> WrapStream<A> for S {
+impl<S: Stream, A: Actor> WrapStream<A> for S {
     type Stream = StreamWrap<S, A>;
     type Item = S::Item;
 
diff --git a/src/fut/stream_finish.rs b/src/fut/stream_finish.rs
index 6a8883063..9294073b8 100644
--- a/src/fut/stream_finish.rs
+++ b/src/fut/stream_finish.rs
@@ -9,34 +9,27 @@ use crate::fut::{ActorFuture, ActorStream};
 /// when stream completes.
 ///
 /// This structure is produced by the `ActorStream::finish` method.
-#[derive(Debug)]
 #[must_use = "streams do nothing unless polled"]
-pub struct StreamFinish<S: ActorStream + Unpin>(S);
+#[pin_project::pin_project]
+#[derive(Debug)]
+pub struct StreamFinish<S: ActorStream>(#[pin] S);
 
-pub fn new<S>(s: S) -> StreamFinish<S>
-where
-    S: ActorStream + Unpin,
-{
+pub fn new<S: ActorStream>(s: S) -> StreamFinish<S> {
     StreamFinish(s)
 }
 
-impl<S: ActorStream> ActorFuture for StreamFinish<S>
-where
-    S: ActorStream + Unpin,
-    Self: Unpin,
-{
+impl<S: ActorStream> ActorFuture for StreamFinish<S> {
     type Output = ();
     type Actor = S::Actor;
 
     fn poll(
-        self: Pin<&mut Self>,
+        mut self: Pin<&mut Self>,
         act: &mut S::Actor,
         ctx: &mut <S::Actor as Actor>::Context,
         task: &mut Context<'_>,
     ) -> Poll<()> {
-        let this = self.get_mut();
         loop {
-            match Pin::new(&mut this.0).poll_next(act, ctx, task) {
+            match self.as_mut().project().0.poll_next(act, ctx, task) {
                 Poll::Pending => return Poll::Pending,
                 Poll::Ready(None) => return Poll::Ready(()),
                 Poll::Ready(Some(_)) => (),
diff --git a/src/fut/stream_fold.rs b/src/fut/stream_fold.rs
index 715a9bfef..1c1759cb2 100644
--- a/src/fut/stream_fold.rs
+++ b/src/fut/stream_fold.rs
@@ -1,4 +1,3 @@
-use std::mem;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
@@ -20,9 +19,11 @@ where
     #[pin]
     stream: S,
     f: F,
+    #[pin]
     state: State<T, Fut::Future>,
 }
 
+#[pin_project(project = FoldStateProj)]
 #[derive(Debug)]
 enum State<T, F>
 where
@@ -32,10 +33,10 @@ where
     Empty,
 
     /// Ready to process the next stream item; current accumulator is the `T`
-    Ready(T),
+    Ready(Option<T>),
 
     /// Working on a future the process the previous stream item
-    Processing(F),
+    Processing(#[pin] F),
 }
 
 pub fn new<S, F, Fut, T>(stream: S, f: F, t: T) -> StreamFold<S, F, Fut, T>
@@ -47,53 +48,53 @@ where
     StreamFold {
         stream,
         f,
-        state: State::Ready(t),
+        state: State::Ready(Some(t)),
     }
 }
 
 impl<S, F, Fut, T> ActorFuture for StreamFold<S, F, Fut, T>
 where
-    S: ActorStream + Unpin,
+    S: ActorStream,
     F: FnMut(T, S::Item, &mut S::Actor, &mut <S::Actor as Actor>::Context) -> Fut,
     Fut: IntoActorFuture<Output = T, Actor = S::Actor>,
-    Fut::Future: ActorFuture + Unpin,
+    Fut::Future: ActorFuture,
 {
     type Output = T;
     type Actor = S::Actor;
 
     fn poll(
-        self: Pin<&mut Self>,
+        mut self: Pin<&mut Self>,
         act: &mut S::Actor,
         ctx: &mut <S::Actor as Actor>::Context,
         task: &mut Context<'_>,
     ) -> Poll<T> {
-        let mut this = self.get_mut();
         loop {
-            match mem::replace(&mut this.state, State::Empty) {
-                State::Empty => panic!("cannot poll Fold twice"),
-                State::Ready(state) => {
-                    match Pin::new(&mut this.stream).poll_next(act, ctx, task) {
+            let this = self.as_mut().project();
+            match this.state.project() {
+                FoldStateProj::Ready(state) => {
+                    match this.stream.poll_next(act, ctx, task) {
                         Poll::Ready(Some(e)) => {
-                            let future = (this.f)(state, e, act, ctx);
+                            let future = (this.f)(state.take().unwrap(), e, act, ctx);
                             let future = future.into_future();
-                            this.state = State::Processing(future);
+                            self.as_mut().project().state.set(State::Processing(future));
                         }
-                        Poll::Ready(None) => return Poll::Ready(state),
-                        Poll::Pending => {
-                            this.state = State::Ready(state);
-                            return Poll::Pending;
+                        Poll::Ready(None) => {
+                            return {
+                                let state = state.take().unwrap();
+                                self.project().state.set(State::Empty);
+                                Poll::Ready(state)
+                            }
                         }
+                        Poll::Pending => return Poll::Pending,
                     }
                 }
-                State::Processing(mut fut) => {
-                    match Pin::new(&mut fut).poll(act, ctx, task) {
-                        Poll::Ready(state) => this.state = State::Ready(state),
-                        Poll::Pending => {
-                            this.state = State::Processing(fut);
-                            return Poll::Pending;
-                        }
+                FoldStateProj::Processing(fut) => match fut.poll(act, ctx, task) {
+                    Poll::Ready(state) => {
+                        self.as_mut().project().state.set(State::Ready(Some(state)))
                     }
-                }
+                    Poll::Pending => return Poll::Pending,
+                },
+                FoldStateProj::Empty => panic!("cannot poll Fold twice"),
             }
         }
     }
diff --git a/src/fut/stream_then.rs b/src/fut/stream_then.rs
index 33f2e2d79..5544d15ad 100644
--- a/src/fut/stream_then.rs
+++ b/src/fut/stream_then.rs
@@ -20,6 +20,7 @@ where
 {
     #[pin]
     stream: S,
+    #[pin]
     future: Option<U::Future>,
     f: F,
 }
@@ -39,36 +40,41 @@ where
 
 impl<S, F: 'static, U> ActorStream for StreamThen<S, F, U>
 where
-    S: ActorStream + Unpin,
+    S: ActorStream,
     F: FnMut(S::Item, &mut S::Actor, &mut <S::Actor as Actor>::Context) -> U,
     U: IntoActorFuture<Actor = S::Actor>,
-    U::Future: Unpin,
 {
     type Item = U::Output;
     type Actor = S::Actor;
 
     fn poll_next(
-        self: Pin<&mut Self>,
+        mut self: Pin<&mut Self>,
         act: &mut S::Actor,
         ctx: &mut <S::Actor as Actor>::Context,
         task: &mut Context<'_>,
     ) -> Poll<Option<U::Output>> {
-        let mut this = self.get_mut();
-        if this.future.is_none() {
-            let item = match Pin::new(&mut this.stream).poll_next(act, ctx, task) {
-                Poll::Pending => return Poll::Pending,
-                Poll::Ready(None) => return Poll::Ready(None),
-                Poll::Ready(Some(e)) => e,
-            };
-            this.future = Some((this.f)(item, act, ctx).into_future());
-        }
-        assert!(this.future.is_some());
-        match Pin::new(this.future.as_mut().unwrap()).poll(act, ctx, task) {
-            Poll::Ready(e) => {
-                this.future = None;
-                Poll::Ready(Some(e))
+        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,
+                    }
+                }
             }
-            Poll::Pending => Poll::Pending,
         }
     }
 }
diff --git a/src/fut/stream_timeout.rs b/src/fut/stream_timeout.rs
index fbe662507..a07248ee9 100644
--- a/src/fut/stream_timeout.rs
+++ b/src/fut/stream_timeout.rs
@@ -1,29 +1,33 @@
 use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
+use std::time::Duration;
 
 use crate::actor::Actor;
-use crate::clock::{self, Delay, Duration};
+use crate::clock::{self, Sleep};
 use crate::fut::ActorStream;
 
 /// Future for the `timeout` combinator, interrupts computations if it takes
 /// more than `timeout`.
 ///
 /// This is created by the `ActorFuture::timeout()` method.
+#[pin_project::pin_project]
 #[derive(Debug)]
 #[must_use = "streams do nothing unless polled"]
 pub struct StreamTimeout<S>
 where
-    S: ActorStream + Unpin,
+    S: ActorStream,
 {
+    #[pin]
     stream: S,
     dur: Duration,
-    timeout: Option<Delay>,
+    #[pin]
+    timeout: Option<Sleep>,
 }
 
 pub fn new<S>(stream: S, timeout: Duration) -> StreamTimeout<S>
 where
-    S: ActorStream + Unpin,
+    S: ActorStream,
 {
     StreamTimeout {
         stream,
@@ -34,7 +38,7 @@ where
 
 impl<S> ActorStream for StreamTimeout<S>
 where
-    S: ActorStream + Unpin,
+    S: ActorStream,
 {
     type Item = Result<S::Item, ()>;
     type Actor = S::Actor;
@@ -45,11 +49,11 @@ where
         ctx: &mut <S::Actor as Actor>::Context,
         task: &mut Context<'_>,
     ) -> Poll<Option<Result<S::Item, ()>>> {
-        let this = self.get_mut();
+        let mut this = self.project();
 
-        match Pin::new(&mut this.stream).poll_next(act, ctx, task) {
+        match this.stream.poll_next(act, ctx, task) {
             Poll::Ready(Some(res)) => {
-                this.timeout.take();
+                this.timeout.set(None);
                 return Poll::Ready(Some(Ok(res)));
             }
             Poll::Ready(None) => return Poll::Ready(None),
@@ -57,15 +61,21 @@ where
         }
 
         if this.timeout.is_none() {
-            this.timeout = Some(clock::delay_for(this.dur));
+            this.timeout.set(Some(clock::sleep(*this.dur)));
         }
 
         // check timeout
-        match Pin::new(this.timeout.as_mut().unwrap()).poll(task) {
-            Poll::Ready(_) => (),
-            Poll::Pending => return Poll::Pending,
+        if this
+            .timeout
+            .as_mut()
+            .as_pin_mut()
+            .unwrap()
+            .poll(task)
+            .is_pending()
+        {
+            return Poll::Pending;
         }
-        this.timeout.take();
+        this.timeout.set(None);
 
         Poll::Ready(Some(Err(())))
     }
diff --git a/src/fut/timeout.rs b/src/fut/timeout.rs
index ba4ba4f29..82ed2e4ad 100644
--- a/src/fut/timeout.rs
+++ b/src/fut/timeout.rs
@@ -6,7 +6,7 @@ use std::time::Duration;
 use pin_project::pin_project;
 
 use crate::actor::Actor;
-use crate::clock::{delay_for, Delay};
+use crate::clock::{sleep, Sleep};
 use crate::fut::ActorFuture;
 
 /// Future for the `timeout` combinator, interrupts computations if it takes
@@ -23,7 +23,7 @@ where
     #[pin]
     fut: F,
     #[pin]
-    timeout: Delay,
+    timeout: Sleep,
 }
 
 pub fn new<F>(future: F, timeout: Duration) -> Timeout<F>
@@ -32,7 +32,7 @@ where
 {
     Timeout {
         fut: future,
-        timeout: delay_for(timeout),
+        timeout: sleep(timeout),
     }
 }
 
diff --git a/src/handler.rs b/src/handler.rs
index fc0888b2c..9eb0fef90 100644
--- a/src/handler.rs
+++ b/src/handler.rs
@@ -102,7 +102,7 @@ pub struct MessageResult<M: Message>(pub M::Result);
 /// even if there are multiple `Msg` sent to `MyActor`.
 /// ```rust, no_run
 /// # use actix::prelude::*;
-/// # use actix::clock::delay_for;
+/// # use actix::clock::sleep;
 /// # use std::time::Duration;
 /// #
 /// # #[derive(Message)]
@@ -126,7 +126,7 @@ pub struct MessageResult<M: Message>(pub M::Result);
 ///                     this.0 = 30;
 ///                 })
 ///                 .then(|_, this, _| {
-///                     delay_for(Duration::from_secs(3)).into_actor(this)
+///                     sleep(Duration::from_secs(3)).into_actor(this)
 ///                 })
 ///                 .map(|_, this, _| {
 ///                     this.0 -= 1;
diff --git a/src/lib.rs b/src/lib.rs
index 09f19254c..67b7286e8 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -42,6 +42,8 @@
 #![allow(clippy::needless_doctest_main)]
 #![deny(nonstandard_style, rust_2018_idioms)]
 #![warn(deprecated_in_future, trivial_casts, trivial_numeric_casts)]
+// TODO: temporary allow deprecated until resolver actor is removed.
+#![allow(deprecated)]
 
 #[doc(hidden)]
 #[cfg(feature = "derive")]
@@ -174,11 +176,11 @@ pub mod dev {
 ///
 /// ```
 /// use std::time::{Duration, Instant};
-/// use tokio::time::delay_for;
+/// use tokio::time::sleep;
 ///
 /// fn main() {
 ///   actix::run(async move {
-///       delay_for(Duration::from_millis(100)).await;
+///       sleep(Duration::from_millis(100)).await;
 ///       actix::System::current().stop();
 ///   });
 /// }
diff --git a/src/utils.rs b/src/utils.rs
index 2c662183d..1a12e0dee 100644
--- a/src/utils.rs
+++ b/src/utils.rs
@@ -6,7 +6,7 @@ use std::time::Duration;
 use futures_channel::oneshot;
 
 use crate::actor::Actor;
-use crate::clock::{interval_at, Delay, Instant, Interval};
+use crate::clock::{interval_at, Instant, Interval, Sleep};
 use crate::fut::{ActorFuture, ActorStream};
 
 pub struct Condition<T>
@@ -80,12 +80,14 @@ where
 /// # }
 
 #[must_use = "future do nothing unless polled"]
+#[pin_project::pin_project]
 pub struct TimerFunc<A>
 where
     A: Actor,
 {
     f: Option<Box<dyn TimerFuncBox<A>>>,
-    timeout: Delay,
+    #[pin]
+    timeout: Sleep,
 }
 
 impl<A> TimerFunc<A>
@@ -99,7 +101,7 @@ where
     {
         TimerFunc {
             f: Some(Box::new(f)),
-            timeout: tokio::time::delay_for(timeout),
+            timeout: tokio::time::sleep(timeout),
         }
     }
 }
@@ -128,8 +130,8 @@ where
         ctx: &mut <Self::Actor as Actor>::Context,
         task: &mut Context<'_>,
     ) -> Poll<Self::Output> {
-        let this = self.get_mut();
-        match Pin::new(&mut this.timeout).poll(task) {
+        let this = self.project();
+        match this.timeout.poll(task) {
             Poll::Ready(_) => {
                 if let Some(f) = this.f.take() {
                     f.call(act, ctx);


Test patch:
diff --git a/tests/test_actor.rs b/tests/test_actor.rs
index 4d403b145..411d8a436 100644
--- a/tests/test_actor.rs
+++ b/tests/test_actor.rs
@@ -3,7 +3,7 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::Arc;
 
 use actix::prelude::*;
-use tokio::time::{delay_for, Duration, Instant};
+use tokio::time::{sleep, Duration, Instant};
 
 #[derive(Clone, Debug)]
 struct Num(usize);
@@ -47,7 +47,7 @@ async fn test_stream() {
         MyActor(act_count, act_err, Running::Stop)
     });
 
-    delay_for(Duration::new(0, 1_000_000)).await;
+    sleep(Duration::new(0, 1_000_000)).await;
 
     assert_eq!(count.load(Ordering::Relaxed), 7);
     assert!(err.load(Ordering::Relaxed));
@@ -97,7 +97,7 @@ async fn test_infinite_stream() {
         StopOnRequest(act_count, act_stopped, act_finished)
     });
 
-    delay_for(Duration::new(0, 1_000_000)).await;
+    sleep(Duration::new(0, 1_000_000)).await;
 
     addr.send(Stop).await.unwrap();
 
diff --git a/tests/test_actors.rs b/tests/test_actors.rs
index e7870bbd9..752b5f930 100644
--- a/tests/test_actors.rs
+++ b/tests/test_actors.rs
@@ -1,3 +1,5 @@
+#![allow(deprecated)]
+
 use actix::actors::resolver;
 use actix::prelude::*;
 
diff --git a/tests/test_address.rs b/tests/test_address.rs
index c335b76c2..c749e90ad 100644
--- a/tests/test_address.rs
+++ b/tests/test_address.rs
@@ -4,7 +4,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
 use actix::prelude::*;
-use tokio::time::{delay_for, Duration};
+use tokio::time::{sleep, Duration};
 
 #[derive(Debug)]
 struct Ping(usize);
@@ -196,7 +196,7 @@ impl Handler<Ping> for TimeoutActor {
     type Result = ();
 
     fn handle(&mut self, _: Ping, ctx: &mut Self::Context) {
-        delay_for(Duration::new(0, 5_000_000))
+        sleep(Duration::new(0, 5_000_000))
             .into_actor(self)
             .wait(ctx);
     }
diff --git a/tests/test_atomic_response.rs b/tests/test_atomic_response.rs
index fe7171fe6..6ad613ae9 100644
--- a/tests/test_atomic_response.rs
+++ b/tests/test_atomic_response.rs
@@ -1,4 +1,4 @@
-use actix::clock::delay_for;
+use actix::clock::sleep;
 use actix::prelude::*;
 use futures_util::stream::StreamExt;
 use std::time::{Duration, Instant};
@@ -18,7 +18,7 @@ impl Handler<Num> for MyActor {
 
     fn handle(&mut self, msg: Num, _: &mut Self::Context) -> Self::Result {
         AtomicResponse::new(Box::pin(
-            delay_for(Duration::from_millis(msg.0 as u64))
+            sleep(Duration::from_millis(msg.0 as u64))
                 .into_actor(self)
                 .map(move |_res, this, _| {
                     this.0 += msg.0;
diff --git a/tests/test_connected.rs b/tests/test_connected.rs
index 849859bd1..13823e817 100644
--- a/tests/test_connected.rs
+++ b/tests/test_connected.rs
@@ -1,6 +1,6 @@
 use actix::prelude::*;
 use std::time::Duration;
-use tokio::time::delay_for;
+use tokio::time::sleep;
 
 struct MyActor;
 
@@ -24,7 +24,7 @@ impl Actor for MyActor {
 async fn test_connected() {
     Arbiter::spawn(async move {
         let addr = MyActor::start(MyActor);
-        delay_for(Duration::from_millis(350)).await;
+        sleep(Duration::from_millis(350)).await;
         drop(addr);
     });
 }
diff --git a/tests/test_context.rs b/tests/test_context.rs
index ac1808086..489c7c015 100644
--- a/tests/test_context.rs
+++ b/tests/test_context.rs
@@ -1,12 +1,13 @@
 #![allow(clippy::let_unit_value)]
+use std::pin::Pin;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
+use std::task::{Context as StdContext, Poll};
 
 use actix::prelude::*;
 use futures_channel::mpsc::unbounded;
 use futures_util::stream::once;
-use futures_util::stream::StreamExt;
-use tokio::time::{delay_for, interval_at, Duration, Instant};
+use tokio::time::{interval_at, sleep, Duration, Instant};
 
 #[derive(Debug, PartialEq)]
 enum Op {
@@ -89,7 +90,7 @@ fn test_add_timeout_cancel() {
         let _addr = MyActor { op: Op::Cancel }.start();
 
         actix_rt::spawn(async move {
-            delay_for(Duration::new(0, 1000)).await;
+            sleep(Duration::new(0, 1000)).await;
             System::current().stop();
         });
     })
@@ -148,7 +149,7 @@ impl Handler<Ping> for ContextWait {
         let cnt = self.cnt.load(Ordering::Relaxed);
         self.cnt.store(cnt + 1, Ordering::Relaxed);
 
-        let fut = delay_for(Duration::from_secs(1));
+        let fut = sleep(Duration::from_secs(1));
         fut.into_actor(self).wait(ctx);
 
         System::current().stop();
@@ -242,7 +243,7 @@ async fn test_nowait_context() {
         addr.do_send(Ping);
     });
 
-    delay_for(Duration::from_millis(200)).await;
+    sleep(Duration::from_millis(200)).await;
 
     assert_eq!(m.load(Ordering::Relaxed), 3);
 }
@@ -264,7 +265,7 @@ async fn test_message_stream_nowait_context() {
         });
     });
 
-    delay_for(Duration::from_millis(200)).await;
+    sleep(Duration::from_millis(200)).await;
 
     assert_eq!(m.load(Ordering::Relaxed), 3);
 }
@@ -286,7 +287,7 @@ fn test_stream_nowait_context() {
         });
 
         actix_rt::spawn(async move {
-            delay_for(Duration::from_millis(200)).await;
+            sleep(Duration::from_millis(200)).await;
             System::current().stop();
         });
     })
@@ -311,7 +312,7 @@ fn test_notify() {
         addr.do_send(Ping);
 
         actix_rt::spawn(async move {
-            delay_for(Duration::from_millis(200)).await;
+            sleep(Duration::from_millis(200)).await;
             System::current().stop();
         });
     })
@@ -400,11 +401,28 @@ impl StreamHandler<CancelPacket> for CancelHandler {
 #[test]
 fn test_cancel_handler() {
     actix::System::run(|| {
+        struct WtfStream {
+            interval: tokio::time::Interval,
+        }
+
+        impl Stream for WtfStream {
+            type Item = CancelPacket;
+
+            fn poll_next(
+                self: Pin<&mut Self>,
+                cx: &mut StdContext<'_>,
+            ) -> Poll<Option<Self::Item>> {
+                self.get_mut()
+                    .interval
+                    .poll_tick(cx)
+                    .map(|_| Some(CancelPacket))
+            }
+        }
+
         CancelHandler::create(|ctx| CancelHandler {
-            source: ctx.add_stream(
-                interval_at(Instant::now(), Duration::from_millis(1))
-                    .map(|_| CancelPacket),
-            ),
+            source: ctx.add_stream(WtfStream {
+                interval: interval_at(Instant::now(), Duration::from_millis(1)),
+            }),
         });
     })
     .unwrap();
@@ -445,14 +463,14 @@ fn test_cancel_completed_with_no_context_item() {
 
         // then, cancel the future which would already be completed
         actix_rt::spawn(async move {
-            delay_for(Duration::from_millis(100)).await;
+            sleep(Duration::from_millis(100)).await;
             addr.do_send(CancelMessage);
         });
 
         // finally, terminate the actor, which shouldn't be blocked unless
         // the actor context ate up CPU time
         actix_rt::spawn(async {
-            delay_for(Duration::from_millis(200)).await;
+            sleep(Duration::from_millis(200)).await;
             System::current().stop();
         });
     })
diff --git a/tests/test_fut.rs b/tests/test_fut.rs
index 7732081a6..bf2f049bd 100644
--- a/tests/test_fut.rs
+++ b/tests/test_fut.rs
@@ -1,7 +1,8 @@
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 
-use actix::clock::{delay_for, Duration};
+use actix::clock::sleep;
 use actix::prelude::*;
 use futures_util::future::FutureExt;
 
@@ -13,7 +14,7 @@ impl Actor for MyActor {
     type Context = actix::Context<Self>;
 
     fn started(&mut self, ctx: &mut Self::Context) {
-        delay_for(Duration::new(0, 5_000_000))
+        sleep(Duration::new(0, 5_000_000))
             .then(|_| async {
                 System::current().stop();
             })
@@ -51,8 +52,8 @@ impl Actor for MyStreamActor {
 
     fn started(&mut self, ctx: &mut Self::Context) {
         let mut s = futures_util::stream::FuturesOrdered::new();
-        s.push(delay_for(Duration::new(0, 5_000_000)));
-        s.push(delay_for(Duration::new(0, 5_000_000)));
+        s.push(sleep(Duration::new(0, 5_000_000)));
+        s.push(sleep(Duration::new(0, 5_000_000)));
 
         s.into_actor(self)
             .timeout(Duration::new(0, 1000))
@@ -60,7 +61,7 @@ impl Actor for MyStreamActor {
                 // Additional waiting time to test `then` call as well
                 Box::pin(
                     async move {
-                        delay_for(Duration::from_millis(500)).await;
+                        sleep(Duration::from_millis(500)).await;
                         res
                     }
                     .into_actor(act),
diff --git a/tests/test_lifecycle.rs b/tests/test_lifecycle.rs
index 6dd73c585..cb06af1e9 100644
--- a/tests/test_lifecycle.rs
+++ b/tests/test_lifecycle.rs
@@ -6,7 +6,7 @@ use std::sync::{Arc, Mutex};
 use actix::prelude::*;
 use futures_channel::oneshot::{channel, Sender};
 use futures_util::future::FutureExt;
-use tokio::time::{delay_for, Duration};
+use tokio::time::{sleep, Duration};
 
 struct MyActor {
     started: Arc<AtomicBool>,
@@ -122,9 +122,9 @@ fn test_stop_after_drop_address() {
         .start();
 
         actix_rt::spawn(async move {
-            delay_for(Duration::new(0, 100)).await;
+            sleep(Duration::new(0, 100)).await;
             drop(addr);
-            delay_for(Duration::new(0, 10_000)).await;
+            sleep(Duration::new(0, 10_000)).await;
             System::current().stop();
         });
     })
@@ -155,7 +155,7 @@ fn test_stop_after_drop_sync_address() {
         .start();
 
         actix_rt::spawn(async move {
-            delay_for(Duration::new(0, 100)).await;
+            sleep(Duration::new(0, 100)).await;
             drop(addr);
             System::current().stop();
         });
@@ -190,13 +190,13 @@ fn test_stop_after_drop_sync_actor() {
         });
 
         actix_rt::spawn(async move {
-            delay_for(Duration::from_secs(2)).await;
+            sleep(Duration::from_secs(2)).await;
             assert!(started2.load(Ordering::Relaxed), "Not started");
             assert!(!stopping2.load(Ordering::Relaxed), "Stopping");
             assert!(!stopped2.load(Ordering::Relaxed), "Stopped");
             drop(addr);
 
-            delay_for(Duration::from_secs(2)).await;
+            sleep(Duration::from_secs(2)).await;
             System::current().stop();
         });
     })
@@ -227,7 +227,7 @@ fn test_stop() {
         .start();
 
         actix_rt::spawn(async move {
-            delay_for(Duration::new(0, 100)).await;
+            sleep(Duration::new(0, 100)).await;
             System::current().stop();
         });
     })
@@ -258,7 +258,7 @@ fn test_stop_restore_after_stopping() {
         .start();
 
         actix_rt::spawn(async move {
-            delay_for(Duration::new(0, 100)).await;
+            sleep(Duration::new(0, 100)).await;
             System::current().stop();
         });
     })
diff --git a/tests/test_sink.rs b/tests/test_sink.rs
index b09223df5..b5257fdde 100644
--- a/tests/test_sink.rs
+++ b/tests/test_sink.rs
@@ -19,11 +19,6 @@ struct MySink {
 impl Sink<Bytes> for MySink {
     type Error = ();
 
-    fn start_send(self: Pin<&mut Self>, bytes: Bytes) -> Result<(), Self::Error> {
-        self.get_mut().queue.push(bytes);
-        Ok(())
-    }
-
     fn poll_ready(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
@@ -51,6 +46,11 @@ impl Sink<Bytes> for MySink {
         }
     }
 
+    fn start_send(self: Pin<&mut Self>, bytes: Bytes) -> Result<(), Self::Error> {
+        self.get_mut().queue.push(bytes);
+        Ok(())
+    }
+
     fn poll_flush(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
diff --git a/tests/test_supervisor.rs b/tests/test_supervisor.rs
index cc4046621..1ac0ac13d 100644
--- a/tests/test_supervisor.rs
+++ b/tests/test_supervisor.rs
@@ -2,7 +2,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
 
 use actix::prelude::*;
-use tokio::time::{delay_for, Duration};
+use tokio::time::{sleep, Duration};
 
 struct Die;
 
@@ -54,7 +54,7 @@ fn test_supervisor_restart() {
         *addr2.lock().unwrap() = Some(addr);
 
         actix::spawn(async move {
-            delay_for(Duration::new(0, 100_000)).await;
+            sleep(Duration::new(0, 100_000)).await;
             System::current().stop();
         });
     })