prepare server release 2.0.0-beta.4

Co-authored-by: Rob Ede <robjtede@icloud.com>

.cargo/config.toml

@@ -0,0 +1,3 @@
+[alias]
+chk = "hack check --workspace --all-features --tests --examples"
+lint = "hack --clean-per-run clippy --workspace --tests --examples"

.github/workflows/ci.yml

@@ -25,6 +25,9 @@ jobs:
     name: ${{ matrix.target.name }} / ${{ matrix.version }}
     runs-on: ${{ matrix.target.os }}
 
+    env:
+      VCPKGRS_DYNAMIC: 1
+
     steps:
       - name: Setup Routing
         if: matrix.target.os == 'macos-latest'
@@ -32,6 +35,17 @@ jobs:
 
       - uses: actions/checkout@v2
 
+      # install OpenSSL on Windows
+      - name: Set vcpkg root
+        if: matrix.target.triple == 'x86_64-pc-windows-msvc' || matrix.target.triple == 'i686-pc-windows-msvc'
+        run: echo "VCPKG_ROOT=$env:VCPKG_INSTALLATION_ROOT" | Out-File -FilePath $env:GITHUB_ENV -Append
+      - name: Install OpenSSL
+        if: matrix.target.triple == 'x86_64-pc-windows-msvc'
+        run: vcpkg install openssl:x64-windows
+      - name: Install OpenSSL
+        if: matrix.target.triple == 'i686-pc-windows-msvc'
+        run: vcpkg install openssl:x86-windows
+
       - name: Install ${{ matrix.version }}
         uses: actions-rs/toolchain@v1
         with:

Cargo.toml

@@ -10,6 +10,8 @@ members = [
   "actix-tracing",
   "actix-utils",
   "bytestring",
+  "local-channel",
+  "local-waker",
 ]
 
 [patch.crates-io]
@@ -23,3 +25,5 @@ actix-tls = { path = "actix-tls" }
 actix-tracing = { path = "actix-tracing" }
 actix-utils = { path = "actix-utils" }
 bytestring = { path = "bytestring" }
+local-channel = { path = "local-channel" }
+local-waker = { path = "local-waker" }

actix-macros/Cargo.toml

@@ -19,5 +19,5 @@ syn = { version = "^1", features = ["full"] }
 [dev-dependencies]
 actix-rt = "2.0.0"
 
-futures-util = { version = "0.3", default-features = false }
+futures-util = { version = "0.3.7", default-features = false }
 trybuild = "1"

actix-router/src/url.rs

@@ -170,13 +170,11 @@ impl Quoter {
             idx += 1;
         }
 
-        if let Some(data) = cloned {
+        cloned.map(|data| {
-            // Unsafe: we get data from http::Uri, which does utf-8 checks already
+            // SAFETY: we get data from http::Uri, which does UTF-8 checks already
             // this code only decodes valid pct encoded values
-            Some(unsafe { String::from_utf8_unchecked(data) })
+            unsafe { String::from_utf8_unchecked(data) }
-        } else {
+        })
-            None
-        }
     }
 }
 

actix-rt/CHANGES.md

@@ -3,6 +3,14 @@
 ## Unreleased - 2021-xx-xx
 
 
+## 2.2.0 - 2021-03-29
+* **BREAKING** `ActixStream::{poll_read_ready, poll_write_ready}` methods now return
+  `Ready` object in ok variant. [#293]
+  * Breakage is acceptable since `ActixStream` was not intended to be public.
+
+[#293] https://github.com/actix/actix-net/pull/293
+
+
 ## 2.1.0 - 2021-02-24
 * Add `ActixStream` extension trait to include readiness methods. [#276]
 * Re-export `tokio::net::TcpSocket` in `net` module [#282]
@@ -64,10 +72,7 @@
 
 
 ## 2.0.0-beta.1 - 2020-12-28
-### Added
 * Add `System::attach_to_tokio` method. [#173]
-
-### Changed
 * Update `tokio` dependency to `1.0`. [#236]
 * Rename `time` module `delay_for` to `sleep`, `delay_until` to `sleep_until`, `Delay` to `Sleep`
   to stay aligned with Tokio's naming. [#236]
@@ -75,27 +80,19 @@
   * These methods now accept `&self` when calling. [#236]
 * Remove `'static` lifetime requirement for `System::run` and `Builder::run`. [#236]
 * `Arbiter::spawn` now panics when `System` is not in scope. [#207]
-
-### Fixed
 * Fix work load issue by removing `PENDING` thread local. [#207]
 
 [#207]: https://github.com/actix/actix-net/pull/207
 [#236]: https://github.com/actix/actix-net/pull/236
 
-## [1.1.1] - 2020-04-30
-
-### Fixed
 
+## 1.1.1 - 2020-04-30
 * Fix memory leak due to [#94] (see [#129] for more detail)
 
 [#129]: https://github.com/actix/actix-net/issues/129
 
-## [1.1.0] - 2020-04-08
-
-**This version has been yanked.**
-
-### Added
 
+## 1.1.0 - 2020-04-08 (YANKED)
 * Expose `System::is_set` to check if current system has ben started [#99]
 * Add `Arbiter::is_running` to check if event loop is running [#124]
 * Add `Arbiter::local_join` associated function
@@ -105,96 +102,57 @@
 [#99]: https://github.com/actix/actix-net/pull/99
 [#124]: https://github.com/actix/actix-net/pull/124
 
-## [1.0.0] - 2019-12-11
 
+## 1.0.0 - 2019-12-11
 * Update dependencies
 
-## [1.0.0-alpha.3] - 2019-12-07
-
-### Fixed
 
+## 1.0.0-alpha.3 - 2019-12-07
+* Migrate to tokio 0.2
 * Fix compilation on non-unix platforms
 
-### Changed
-
-* Migrate to tokio 0.2
-
-
-## [1.0.0-alpha.2] - 2019-12-02
-
-Added
 
+## 1.0.0-alpha.2 - 2019-12-02
 * Export `main` and `test` attribute macros
-
 * Export `time` module (re-export of tokio-timer)
-
 * Export `net` module (re-export of tokio-net)
 
 
-## [1.0.0-alpha.1] - 2019-11-22
+## 1.0.0-alpha.1 - 2019-11-22
-
-### Changed
-
 * Migrate to std::future and tokio 0.2
 
 
-## [0.2.6] - 2019-11-14
+## 0.2.6 - 2019-11-14
-
+* Allow to join arbiter's thread. #60
-### Fixed
-
 * Fix arbiter's thread panic message.
 
-### Added
-
-* Allow to join arbiter's thread. #60
-
-
-## [0.2.5] - 2019-09-02
-
-### Added
 
+## 0.2.5 - 2019-09-02
 * Add arbiter specific storage
 
 
-## [0.2.4] - 2019-07-17
+## 0.2.4 - 2019-07-17
-
-### Changed
-
 * Avoid a copy of the Future when initializing the Box. #29
 
 
-## [0.2.3] - 2019-06-22
+## 0.2.3 - 2019-06-22
-
+* Allow to start System using existing CurrentThread Handle #22
-### Added
-
-* Allow to start System using exsiting CurrentThread Handle #22
 
 
-## [0.2.2] - 2019-03-28
+## 0.2.2 - 2019-03-28
-
-### Changed
-
 * Moved `blocking` module to `actix-threadpool` crate
 
 
-## [0.2.1] - 2019-03-11
+## 0.2.1 - 2019-03-11
-
-### Added
-
 * Added `blocking` module
-
+* Added `Arbiter::exec_fn` - execute fn on the arbiter's thread
-* Arbiter::exec_fn - execute fn on the arbiter's thread
+* Added `Arbiter::exec` - execute fn on the arbiter's thread and wait result
-
-* Arbiter::exec - execute fn on the arbiter's thread and wait result
 
 
-## [0.2.0] - 2019-03-06
+## 0.2.0 - 2019-03-06
-
 * `run` method returns `io::Result<()>`
-
 * Removed `Handle`
 
 
-## [0.1.0] - 2018-12-09
+## 0.1.0 - 2018-12-09
-
 * Initial release

actix-rt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-rt"
-version = "2.1.0"
+version = "2.2.0"
 authors = [
     "Nikolay Kim <fafhrd91@gmail.com>",
     "Rob Ede <robjtede@icloud.com>",
@@ -26,7 +26,7 @@ macros = ["actix-macros"]
 actix-macros = { version = "0.2.0", optional = true }
 
 futures-core = { version = "0.3", default-features = false }
-tokio = { version = "1.2", features = ["rt", "net", "parking_lot", "signal", "sync", "time"] }
+tokio = { version = "1.3", features = ["rt", "net", "parking_lot", "signal", "sync", "time"] }
 
 [dev-dependencies]
 tokio = { version = "1.2", features = ["full"] }

actix-rt/README.md

@@ -3,11 +3,11 @@
 > Tokio-based single-threaded async runtime for the Actix ecosystem.
 
 [![crates.io](https://img.shields.io/crates/v/actix-rt?label=latest)](https://crates.io/crates/actix-rt)
-[![Documentation](https://docs.rs/actix-rt/badge.svg?version=2.1.0)](https://docs.rs/actix-rt/2.1.0)
+[![Documentation](https://docs.rs/actix-rt/badge.svg?version=2.2.0)](https://docs.rs/actix-rt/2.2.0)
 [![Version](https://img.shields.io/badge/rustc-1.46+-ab6000.svg)](https://blog.rust-lang.org/2020/03/12/Rust-1.46.html)
 ![MIT or Apache 2.0 licensed](https://img.shields.io/crates/l/actix-rt.svg)
 <br />
-[![dependency status](https://deps.rs/crate/actix-rt/2.1.0/status.svg)](https://deps.rs/crate/actix-rt/2.1.0)
+[![dependency status](https://deps.rs/crate/actix-rt/2.2.0/status.svg)](https://deps.rs/crate/actix-rt/2.2.0)
 ![Download](https://img.shields.io/crates/d/actix-rt.svg)
 [![Chat on Discord](https://img.shields.io/discord/771444961383153695?label=chat&logo=discord)](https://discord.gg/WghFtEH6Hb)
 

actix-rt/examples/multi_thread_system.rs

@@ -0,0 +1,60 @@
+//! An example on how to build a multi-thread tokio runtime for Actix System.
+//! Then spawn async task that can make use of work stealing of tokio runtime.
+
+use actix_rt::System;
+
+fn main() {
+    System::with_tokio_rt(|| {
+        // build system with a multi-thread tokio runtime.
+        tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(2)
+            .enable_all()
+            .build()
+            .unwrap()
+    })
+    .block_on(async_main());
+}
+
+// async main function that acts like #[actix_web::main] or #[tokio::main]
+async fn async_main() {
+    let (tx, rx) = tokio::sync::oneshot::channel();
+
+    // get a handle to system arbiter and spawn async task on it
+    System::current().arbiter().spawn(async {
+        // use tokio::spawn to get inside the context of multi thread tokio runtime
+        let h1 = tokio::spawn(async {
+            println!("thread id is {:?}", std::thread::current().id());
+            std::thread::sleep(std::time::Duration::from_secs(2));
+        });
+
+        // work stealing occurs for this task spawn
+        let h2 = tokio::spawn(async {
+            println!("thread id is {:?}", std::thread::current().id());
+        });
+
+        h1.await.unwrap();
+        h2.await.unwrap();
+        let _ = tx.send(());
+    });
+
+    rx.await.unwrap();
+
+    let (tx, rx) = tokio::sync::oneshot::channel();
+    let now = std::time::Instant::now();
+
+    // without additional tokio::spawn, all spawned tasks run on single thread
+    System::current().arbiter().spawn(async {
+        println!("thread id is {:?}", std::thread::current().id());
+        std::thread::sleep(std::time::Duration::from_secs(2));
+        let _ = tx.send(());
+    });
+
+    // previous spawn task has blocked the system arbiter thread
+    // so this task will wait for 2 seconds until it can be run
+    System::current().arbiter().spawn(async move {
+        println!("thread id is {:?}", std::thread::current().id());
+        assert!(now.elapsed() > std::time::Duration::from_secs(2));
+    });
+
+    rx.await.unwrap();
+}

actix-rt/src/lib.rs

@@ -72,9 +72,14 @@ pub mod signal {
 pub mod net {
     //! TCP/UDP/Unix bindings (mostly Tokio re-exports).
 
-    use std::task::{Context, Poll};
+    use std::{
+        future::Future,
+        io,
+        task::{Context, Poll},
+    };
 
-    use tokio::io::{AsyncRead, AsyncWrite};
+    pub use tokio::io::Ready;
+    use tokio::io::{AsyncRead, AsyncWrite, Interest};
     pub use tokio::net::UdpSocket;
     pub use tokio::net::{TcpListener, TcpSocket, TcpStream};
 
@@ -82,36 +87,55 @@ pub mod net {
     pub use tokio::net::{UnixDatagram, UnixListener, UnixStream};
 
     /// Extension trait over async read+write types that can also signal readiness.
-    pub trait ActixStream: AsyncRead + AsyncWrite + Unpin + 'static {
+    #[doc(hidden)]
+    pub trait ActixStream: AsyncRead + AsyncWrite + Unpin {
         /// Poll stream and check read readiness of Self.
         ///
         /// See [tokio::net::TcpStream::poll_read_ready] for detail on intended use.
-        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<std::io::Result<()>>;
+        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>>;
 
         /// Poll stream and check write readiness of Self.
         ///
         /// See [tokio::net::TcpStream::poll_write_ready] for detail on intended use.
-        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<std::io::Result<()>>;
+        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>>;
     }
 
     impl ActixStream for TcpStream {
-        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
-            TcpStream::poll_read_ready(self, cx)
+            let ready = self.ready(Interest::READABLE);
+            tokio::pin!(ready);
+            ready.poll(cx)
         }
 
-        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
-            TcpStream::poll_write_ready(self, cx)
+            let ready = self.ready(Interest::WRITABLE);
+            tokio::pin!(ready);
+            ready.poll(cx)
         }
     }
 
     #[cfg(unix)]
     impl ActixStream for UnixStream {
-        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
-            UnixStream::poll_read_ready(self, cx)
+            let ready = self.ready(Interest::READABLE);
+            tokio::pin!(ready);
+            ready.poll(cx)
         }
 
-        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
-            UnixStream::poll_write_ready(self, cx)
+            let ready = self.ready(Interest::WRITABLE);
+            tokio::pin!(ready);
+            ready.poll(cx)
+        }
+    }
+
+    impl<Io: ActixStream + ?Sized> ActixStream for Box<Io> {
+        fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
+            (**self).poll_read_ready(cx)
+        }
+
+        fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
+            (**self).poll_write_ready(cx)
         }
     }
 }

actix-server/CHANGES.md

@@ -3,6 +3,12 @@
 ## Unreleased - 2021-xx-xx
 
 
+## 2.0.0-beta.4 - 2021-04-01
+* Prevent panic when `shutdown_timeout` is very large. [f9262db]
+
+[f9262db]: https://github.com/actix/actix-net/commit/f9262db
+
+
 ## 2.0.0-beta.3 - 2021-02-06
 * Hidden `ServerBuilder::start` method has been removed. Use `ServerBuilder::run`. [#246]
 * Add retry for EINTR signal (`io::Interrupted`) in `Accept`'s poll loop. [#264]

actix-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-server"
-version = "2.0.0-beta.3"
+version = "2.0.0-beta.4"
 authors = [
     "Nikolay Kim <fafhrd91@gmail.com>",
     "fakeshadow <24548779@qq.com>",
@@ -9,7 +9,6 @@ description = "General purpose TCP server built for the Actix ecosystem"
 keywords = ["network", "framework", "async", "futures"]
 homepage = "https://actix.rs"
 repository = "https://github.com/actix/actix-net.git"
-documentation = "https://docs.rs/actix-server"
 categories = ["network-programming", "asynchronous"]
 license = "MIT OR Apache-2.0"
 edition = "2018"
@@ -22,9 +21,8 @@ path = "src/lib.rs"
 default = []
 
 [dependencies]
-actix-codec = "0.4.0-beta.1"
 actix-rt = { version = "2.0.0", default-features = false }
-actix-service = "2.0.0-beta.4"
+actix-service = "2.0.0-beta.5"
 actix-utils = "3.0.0-beta.2"
 
 futures-core = { version = "0.3.7", default-features = false, features = ["alloc"] }
@@ -35,7 +33,9 @@ slab = "0.4"
 tokio = { version = "1.2", features = ["sync"] }
 
 [dev-dependencies]
+actix-codec = "0.4.0-beta.1"
 actix-rt = "2.0.0"
+
 bytes = "1"
 env_logger = "0.8"
 futures-util = { version = "0.3.7", default-features = false, features = ["sink"] }

actix-server/src/accept.rs

@@ -2,7 +2,7 @@ use std::time::Duration;
 use std::{io, thread};
 
 use actix_rt::{
-    time::{sleep_until, Instant},
+    time::{sleep, Instant},
     System,
 };
 use log::{error, info};
@@ -16,14 +16,17 @@ use crate::worker::{Conn, WorkerHandle};
 use crate::Token;
 
 struct ServerSocketInfo {
-    // addr for socket. mainly used for logging.
+    /// Address of socket. Mainly used for logging.
     addr: SocketAddr,
-    // be ware this is the crate token for identify socket and should not be confused with
+
-    // mio::Token
+    /// Beware this is the crate token for identify socket and should not be confused
+    /// with `mio::Token`.
     token: Token,
+
     lst: MioListener,
-    // timeout is used to mark the deadline when this socket's listener should be registered again
+
-    // after an error.
+    /// Timeout is used to mark the deadline when this socket's listener should be registered again
+    /// after an error.
     timeout: Option<Instant>,
 }
 
@@ -184,38 +187,26 @@ impl Accept {
                         let mut guard = self.waker.guard();
                         match guard.pop_front() {
                             // worker notify it becomes available. we may want to recover
-                            // from  backpressure.
+                            // from backpressure.
                             Some(WakerInterest::WorkerAvailable) => {
                                 drop(guard);
                                 self.maybe_backpressure(&mut sockets, false);
                             }
-                            // a new worker thread is made and it's handle would be added
+                            // a new worker thread is made and it's handle would be added to Accept
-                            // to Accept
                             Some(WakerInterest::Worker(handle)) => {
                                 drop(guard);
                                 // maybe we want to recover from a backpressure.
                                 self.maybe_backpressure(&mut sockets, false);
                                 self.handles.push(handle);
                             }
-                            // got timer interest and it's time to try register socket(s)
+                            // got timer interest and it's time to try register socket(s) again
-                            // again.
                             Some(WakerInterest::Timer) => {
                                 drop(guard);
                                 self.process_timer(&mut sockets)
                             }
                             Some(WakerInterest::Pause) => {
                                 drop(guard);
-                                sockets.iter_mut().for_each(|(_, info)| {
+                                self.deregister_all(&mut sockets);
-                                    match self.deregister(info) {
-                                        Ok(_) => info!(
-                                            "Paused accepting connections on {}",
-                                            info.addr
-                                        ),
-                                        Err(e) => {
-                                            error!("Can not deregister server socket {}", e)
-                                        }
-                                    }
-                                });
                             }
                             Some(WakerInterest::Resume) => {
                                 drop(guard);
@@ -226,10 +217,9 @@ impl Accept {
                             Some(WakerInterest::Stop) => {
                                 return self.deregister_all(&mut sockets);
                             }
-                            // waker queue is drained.
+                            // waker queue is drained
                             None => {
-                                // Reset the WakerQueue before break so it does not grow
+                                // Reset the WakerQueue before break so it does not grow infinitely
-                                // infinitely.
                                 WakerQueue::reset(&mut guard);
                                 break 'waker;
                             }
@@ -246,16 +236,23 @@ impl Accept {
 
     fn process_timer(&self, sockets: &mut Slab<ServerSocketInfo>) {
         let now = Instant::now();
-        sockets.iter_mut().for_each(|(token, info)| {
+        sockets
-            // only the ServerSocketInfo have an associate timeout value was de registered.
+            .iter_mut()
-            if let Some(inst) = info.timeout.take() {
+            // Only sockets that had an associated timeout were deregistered.
-                if now > inst {
+            .filter(|(_, info)| info.timeout.is_some())
-                    self.register_logged(token, info);
+            .for_each(|(token, info)| {
-                } else {
+                let inst = info.timeout.take().unwrap();
+
+                if now < inst {
                     info.timeout = Some(inst);
+                } else if !self.backpressure {
+                    self.register_logged(token, info);
                 }
-            }
+
-        });
+                // Drop the timeout if server is in backpressure and socket timeout is expired.
+                // When server recovers from backpressure it will register all sockets without
+                // a timeout value so this socket register will be delayed till then.
+            });
     }
 
     #[cfg(not(target_os = "windows"))]
@@ -293,136 +290,145 @@ impl Accept {
         self.poll.registry().deregister(&mut info.lst)
     }
 
+    fn deregister_logged(&self, info: &mut ServerSocketInfo) {
+        match self.deregister(info) {
+            Ok(_) => info!("Paused accepting connections on {}", info.addr),
+            Err(e) => {
+                error!("Can not deregister server socket {}", e)
+            }
+        }
+    }
+
     fn deregister_all(&self, sockets: &mut Slab<ServerSocketInfo>) {
         sockets.iter_mut().for_each(|(_, info)| {
-            info!("Accepting connections on {} has been paused", info.addr);
+            self.deregister_logged(info);
-            let _ = self.deregister(info);
         });
     }
 
     fn maybe_backpressure(&mut self, sockets: &mut Slab<ServerSocketInfo>, on: bool) {
-        if self.backpressure {
+        // Only operate when server is in a different backpressure than the given flag.
-            if !on {
+        if self.backpressure != on {
+            if on {
+                self.backpressure = true;
+                // TODO: figure out if timing out sockets can be safely de-registered twice.
+                self.deregister_all(sockets);
+            } else {
                 self.backpressure = false;
-                for (token, info) in sockets.iter_mut() {
+                sockets
-                    if info.timeout.is_some() {
+                    .iter_mut()
-                        // socket will attempt to re-register itself when its timeout completes
+                    // Only operate on sockets without associated timeout.
-                        continue;
+                    // Sockets with it will attempt to re-register when their timeout expires.
-                    }
+                    .filter(|(_, info)| info.timeout.is_none())
-                    self.register_logged(token, info);
+                    .for_each(|(token, info)| self.register_logged(token, info));
-                }
             }
-        } else if on {
-            self.backpressure = true;
-            self.deregister_all(sockets);
         }
     }
 
-    fn accept_one(&mut self, sockets: &mut Slab<ServerSocketInfo>, mut msg: Conn) {
+    fn accept_one(&mut self, sockets: &mut Slab<ServerSocketInfo>, mut conn: Conn) {
         if self.backpressure {
+            // send_connection would remove fault worker from handles.
+            // worst case here is conn get dropped after all handles are gone.
             while !self.handles.is_empty() {
-                match self.handles[self.next].send(msg) {
+                match self.send_connection(sockets, conn) {
-                    Ok(_) => {
+                    Ok(_) => return,
-                        self.set_next();
+                    Err(c) => conn = c,
-                        break;
-                    }
-                    Err(tmp) => {
-                        // worker lost contact and could be gone. a message is sent to
-                        // `ServerBuilder` future to notify it a new worker should be made.
-                        // after that remove the fault worker.
-                        self.srv.worker_faulted(self.handles[self.next].idx);
-                        msg = tmp;
-                        self.handles.swap_remove(self.next);
-                        if self.handles.is_empty() {
-                            error!("No workers");
-                            return;
-                        } else if self.handles.len() <= self.next {
-                            self.next = 0;
-                        }
-                        continue;
-                    }
                 }
             }
         } else {
+            // Do one round and try to send conn to all workers until it succeed.
+            // Start from self.next.
             let mut idx = 0;
             while idx < self.handles.len() {
                 idx += 1;
                 if self.handles[self.next].available() {
-                    match self.handles[self.next].send(msg) {
+                    match self.send_connection(sockets, conn) {
-                        Ok(_) => {
+                        Ok(_) => return,
-                            self.set_next();
+                        Err(c) => conn = c,
-                            return;
-                        }
-                        // worker lost contact and could be gone. a message is sent to
-                        // `ServerBuilder` future to notify it a new worker should be made.
-                        // after that remove the fault worker and enter backpressure if necessary.
-                        Err(tmp) => {
-                            self.srv.worker_faulted(self.handles[self.next].idx);
-                            msg = tmp;
-                            self.handles.swap_remove(self.next);
-                            if self.handles.is_empty() {
-                                error!("No workers");
-                                self.maybe_backpressure(sockets, true);
-                                return;
-                            } else if self.handles.len() <= self.next {
-                                self.next = 0;
-                            }
-                            continue;
-                        }
                     }
+                } else {
+                    self.set_next();
                 }
-                self.set_next();
             }
-            // enable backpressure
+            // Sending Conn failed due to either all workers are in error or not available.
+            // Enter backpressure state and try again.
             self.maybe_backpressure(sockets, true);
-            self.accept_one(sockets, msg);
+            self.accept_one(sockets, conn);
         }
     }
 
-    // set next worker handle that would accept work.
+    // Set next worker handle that would accept work.
     fn set_next(&mut self) {
         self.next = (self.next + 1) % self.handles.len();
     }
 
+    // Send connection to worker and handle error.
+    fn send_connection(
+        &mut self,
+        sockets: &mut Slab<ServerSocketInfo>,
+        conn: Conn,
+    ) -> Result<(), Conn> {
+        match self.handles[self.next].send(conn) {
+            Ok(_) => {
+                self.set_next();
+                Ok(())
+            }
+            Err(conn) => {
+                // worker lost contact and could be gone. a message is sent to
+                // `ServerBuilder` future to notify it a new worker should be made.
+                // after that remove the fault worker and enter backpressure if necessary.
+                self.srv.worker_faulted(self.handles[self.next].idx);
+                self.handles.swap_remove(self.next);
+                if self.handles.is_empty() {
+                    error!("No workers");
+                    self.maybe_backpressure(sockets, true);
+                    // All workers are gone and Conn is nowhere to be sent.
+                    // Treat this situation as Ok and drop Conn.
+                    return Ok(());
+                } else if self.handles.len() <= self.next {
+                    self.next = 0;
+                }
+                Err(conn)
+            }
+        }
+    }
+
     fn accept(&mut self, sockets: &mut Slab<ServerSocketInfo>, token: usize) {
         loop {
-            let msg = if let Some(info) = sockets.get_mut(token) {
+            let info = sockets
-                match info.lst.accept() {
+                .get_mut(token)
-                    Ok(Some((io, addr))) => Conn {
+                .expect("ServerSocketInfo is removed from Slab");
+
+            match info.lst.accept() {
+                Ok(io) => {
+                    let msg = Conn {
                         io,
                         token: info.token,
-                        peer: Some(addr),
+                    };
-                    },
+                    self.accept_one(sockets, msg);
-                    Ok(None) => return,
-                    Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => return,
-                    Err(ref e) if connection_error(e) => continue,
-                    Err(e) => {
-                        // deregister listener temporary
-                        error!("Error accepting connection: {}", e);
-                        if let Err(err) = self.deregister(info) {
-                            error!("Can not deregister server socket {}", err);
-                        }
-
-                        // sleep after error. write the timeout to socket info as later the poll
-                        // would need it mark which socket and when it's listener should be
-                        // registered.
-                        info.timeout = Some(Instant::now() + Duration::from_millis(500));
-
-                        // after the sleep a Timer interest is sent to Accept Poll
-                        let waker = self.waker.clone();
-                        System::current().arbiter().spawn(async move {
-                            sleep_until(Instant::now() + Duration::from_millis(510)).await;
-                            waker.wake(WakerInterest::Timer);
-                        });
-
-                        return;
-                    }
                 }
-            } else {
+                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => return,
-                return;
+                Err(ref e) if connection_error(e) => continue,
-            };
+                Err(e) => {
+                    error!("Error accepting connection: {}", e);
 
-            self.accept_one(sockets, msg);
+                    // deregister listener temporary
+                    self.deregister_logged(info);
+
+                    // sleep after error. write the timeout to socket info as later
+                    // the poll would need it mark which socket and when it's
+                    // listener should be registered
+                    info.timeout = Some(Instant::now() + Duration::from_millis(500));
+
+                    // after the sleep a Timer interest is sent to Accept Poll
+                    let waker = self.waker.clone();
+                    System::current().arbiter().spawn(async move {
+                        sleep(Duration::from_millis(510)).await;
+                        waker.wake(WakerInterest::Timer);
+                    });
+
+                    return;
+                }
+            };
         }
     }
 }

actix-server/src/builder.rs

@@ -1,12 +1,12 @@
-use std::future::Future;
+use std::{
-use std::pin::Pin;
+    future::Future,
-use std::task::{Context, Poll};
+    io, mem,
-use std::time::Duration;
+    pin::Pin,
-use std::{io, mem};
+    task::{Context, Poll},
+    time::Duration,
+};
 
-use actix_rt::net::TcpStream;
+use actix_rt::{self as rt, net::TcpStream, time::sleep, System};
-use actix_rt::time::{sleep_until, Instant};
-use actix_rt::{self as rt, System};
 use log::{error, info};
 use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
 use tokio::sync::oneshot;
@@ -122,13 +122,13 @@ impl ServerBuilder {
         self
     }
 
-    /// Stop actix system.
+    /// Stop Actix system.
     pub fn system_exit(mut self) -> Self {
         self.exit = true;
         self
     }
 
-    /// Disable signal handling
+    /// Disable signal handling.
     pub fn disable_signals(mut self) -> Self {
         self.no_signals = true;
         self
@@ -136,9 +136,8 @@ impl ServerBuilder {
 
     /// Timeout for graceful workers shutdown in seconds.
     ///
-    /// After receiving a stop signal, workers have this much time to finish
+    /// After receiving a stop signal, workers have this much time to finish serving requests.
-    /// serving requests. Workers still alive after the timeout are force
+    /// Workers still alive after the timeout are force dropped.
-    /// dropped.
     ///
     /// By default shutdown timeout sets to 30 seconds.
     pub fn shutdown_timeout(mut self, sec: u64) -> Self {
@@ -147,11 +146,10 @@ impl ServerBuilder {
         self
     }
 
-    /// Execute external configuration as part of the server building
+    /// Execute external configuration as part of the server building process.
-    /// process.
     ///
-    /// This function is useful for moving parts of configuration to a
+    /// This function is useful for moving parts of configuration to a different module or
-    /// different module or even library.
+    /// even library.
     pub fn configure<F>(mut self, f: F) -> io::Result<ServerBuilder>
     where
         F: Fn(&mut ServiceConfig) -> io::Result<()>,
@@ -268,6 +266,7 @@ impl ServerBuilder {
 
         self.sockets
             .push((token, name.as_ref().to_string(), MioListener::from(lst)));
+
         Ok(self)
     }
 
@@ -393,7 +392,7 @@ impl ServerBuilder {
                         }
                         if exit {
                             rt::spawn(async {
-                                sleep_until(Instant::now() + Duration::from_millis(300)).await;
+                                sleep(Duration::from_millis(300)).await;
                                 System::current().stop();
                             });
                         }
@@ -402,7 +401,7 @@ impl ServerBuilder {
                     // we need to stop system if server was spawned
                     if self.exit {
                         rt::spawn(async {
-                            sleep_until(Instant::now() + Duration::from_millis(300)).await;
+                            sleep(Duration::from_millis(300)).await;
                             System::current().stop();
                         });
                     }

actix-server/src/socket.rs

@@ -40,15 +40,11 @@ impl MioListener {
         }
     }
 
-    pub(crate) fn accept(&self) -> io::Result<Option<(MioStream, SocketAddr)>> {
+    pub(crate) fn accept(&self) -> io::Result<MioStream> {
         match *self {
-            MioListener::Tcp(ref lst) => lst
+            MioListener::Tcp(ref lst) => lst.accept().map(|(stream, _)| MioStream::Tcp(stream)),
-                .accept()
-                .map(|(stream, addr)| Some((MioStream::Tcp(stream), SocketAddr::Tcp(addr)))),
             #[cfg(unix)]
-            MioListener::Uds(ref lst) => lst
+            MioListener::Uds(ref lst) => lst.accept().map(|(stream, _)| MioStream::Uds(stream)),
-                .accept()
-                .map(|(stream, addr)| Some((MioStream::Uds(stream), SocketAddr::Uds(addr)))),
         }
     }
 }

actix-server/src/test_server.rs

@@ -92,10 +92,10 @@ impl TestServer {
         let port = addr.port();
 
         TestServerRuntime {
-            system,
             addr,
             host,
             port,
+            system,
         }
     }
 

actix-server/src/worker.rs

@@ -5,7 +5,7 @@ use std::sync::Arc;
 use std::task::{Context, Poll};
 use std::time::Duration;
 
-use actix_rt::time::{sleep_until, Instant, Sleep};
+use actix_rt::time::{sleep, Sleep};
 use actix_rt::{spawn, Arbiter};
 use actix_utils::counter::Counter;
 use futures_core::future::LocalBoxFuture;
@@ -14,7 +14,7 @@ use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
 use tokio::sync::oneshot;
 
 use crate::service::{BoxedServerService, InternalServiceFactory};
-use crate::socket::{MioStream, SocketAddr};
+use crate::socket::MioStream;
 use crate::waker_queue::{WakerInterest, WakerQueue};
 use crate::{join_all, Token};
 
@@ -31,7 +31,6 @@ pub(crate) struct StopCommand {
 pub(crate) struct Conn {
     pub io: MioStream,
     pub token: Token,
-    pub peer: Option<SocketAddr>,
 }
 
 static MAX_CONNS: AtomicUsize = AtomicUsize::new(25600);
@@ -361,8 +360,8 @@ impl Future for ServerWorker {
                 if num != 0 {
                     info!("Graceful worker shutdown, {} connections", num);
                     self.state = WorkerState::Shutdown(
-                        Box::pin(sleep_until(Instant::now() + Duration::from_secs(1))),
+                        Box::pin(sleep(Duration::from_secs(1))),
-                        Box::pin(sleep_until(Instant::now() + self.config.shutdown_timeout)),
+                        Box::pin(sleep(self.config.shutdown_timeout)),
                         Some(result),
                     );
                 } else {
@@ -438,7 +437,7 @@ impl Future for ServerWorker {
 
                 // sleep for 1 second and then check again
                 if t1.as_mut().poll(cx).is_ready() {
-                    *t1 = Box::pin(sleep_until(Instant::now() + Duration::from_secs(1)));
+                    *t1 = Box::pin(sleep(Duration::from_secs(1)));
                     let _ = t1.as_mut().poll(cx);
                 }
 

actix-server/tests/test_server.rs

@@ -4,7 +4,8 @@ use std::{net, thread, time};
 
 use actix_server::Server;
 use actix_service::fn_service;
-use futures_util::future::{lazy, ok};
+use actix_utils::future::ok;
+use futures_util::future::lazy;
 
 fn unused_addr() -> net::SocketAddr {
     let addr: net::SocketAddr = "127.0.0.1:0".parse().unwrap();
@@ -30,6 +31,7 @@ fn test_bind() {
                 .unwrap()
                 .run()
         }));
+
         let _ = tx.send((srv, actix_rt::System::current()));
         let _ = sys.run();
     });
@@ -175,6 +177,7 @@ fn test_configure() {
                 .workers(1)
                 .run()
         }));
+
         let _ = tx.send((srv, actix_rt::System::current()));
         let _ = sys.run();
     });

actix-service/CHANGES.md

@@ -3,6 +3,14 @@
 ## Unreleased - 2021-xx-xx
 
 
+## 2.0.0-beta.5 - 2021-03-15
+* Add default `Service` trait impl for `Rc<S: Service>` and `&S: Service`. [#288]
+* Add `boxed::rc_service` function for constructing `boxed::RcService` type [#290]
+
+[#288]: https://github.com/actix/actix-net/pull/288
+[#290]: https://github.com/actix/actix-net/pull/290
+
+
 ## 2.0.0-beta.4 - 2021-02-04
 * `Service::poll_ready` and `Service::call` receive `&self`. [#247]
 * `apply_fn` and `apply_fn_factory` now receive `Fn(Req, &Service)` function type. [#247]

actix-service/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-service"
-version = "2.0.0-beta.4"
+version = "2.0.0-beta.5"
 authors = [
     "Nikolay Kim <fafhrd91@gmail.com>",
     "Rob Ede <robjtede@icloud.com>",

actix-service/README.md

@@ -3,11 +3,11 @@
 > Service trait and combinators for representing asynchronous request/response operations.
 
 [![crates.io](https://img.shields.io/crates/v/actix-service?label=latest)](https://crates.io/crates/actix-service)
-[![Documentation](https://docs.rs/actix-service/badge.svg?version=2.0.0-beta.4)](https://docs.rs/actix-service/2.0.0-beta.4)
+[![Documentation](https://docs.rs/actix-service/badge.svg?version=2.0.0-beta.5)](https://docs.rs/actix-service/2.0.0-beta.5)
 [![Version](https://img.shields.io/badge/rustc-1.46+-ab6000.svg)](https://blog.rust-lang.org/2020/03/12/Rust-1.46.html)
 ![License](https://img.shields.io/crates/l/actix-service.svg)
-[![Dependency Status](https://deps.rs/crate/actix-service/2.0.0-beta.4/status.svg)](https://deps.rs/crate/actix-service/2.0.0-beta.4)
+[![Dependency Status](https://deps.rs/crate/actix-service/2.0.0-beta.5/status.svg)](https://deps.rs/crate/actix-service/2.0.0-beta.5)
-[![Download](https://img.shields.io/crates/d/actix-service.svg)](https://crates.io/crates/actix-service)
+![Download](https://img.shields.io/crates/d/actix-service.svg)
 [![Chat on Discord](https://img.shields.io/discord/771444961383153695?label=chat&logo=discord)](https://discord.gg/NWpN5mmg3x)
 
 See documentation for detailed explanations of these components: https://docs.rs/actix-service.

actix-service/src/boxed.rs

@@ -1,21 +1,65 @@
-use alloc::boxed::Box;
+//! Trait object forms of services and service factories.
-use core::{
+
-    future::Future,
+use alloc::{boxed::Box, rc::Rc};
-    marker::PhantomData,
+use core::{future::Future, pin::Pin};
-    pin::Pin,
-    task::{Context, Poll},
-};
 
 use crate::{Service, ServiceFactory};
 
+/// A boxed future without a Send bound or lifetime parameters.
 pub type BoxFuture<T> = Pin<Box<dyn Future<Output = T>>>;
 
-pub type BoxService<Req, Res, Err> =
+macro_rules! service_object {
-    Box<dyn Service<Req, Response = Res, Error = Err, Future = BoxFuture<Result<Res, Err>>>>;
+    ($name: ident, $type: tt, $fn_name: ident) => {
+        /// Type alias for service trait object.
+        pub type $name<Req, Res, Err> = $type<
+            dyn Service<Req, Response = Res, Error = Err, Future = BoxFuture<Result<Res, Err>>>,
+        >;
 
+        /// Create service trait object.
+        pub fn $fn_name<S, Req>(service: S) -> $name<Req, S::Response, S::Error>
+        where
+            S: Service<Req> + 'static,
+            Req: 'static,
+            S::Future: 'static,
+        {
+            $type::new(ServiceWrapper::new(service))
+        }
+    };
+}
+
+service_object!(BoxService, Box, service);
+service_object!(RcService, Rc, rc_service);
+
+struct ServiceWrapper<S> {
+    inner: S,
+}
+
+impl<S> ServiceWrapper<S> {
+    fn new(inner: S) -> Self {
+        Self { inner }
+    }
+}
+
+impl<S, Req, Res, Err> Service<Req> for ServiceWrapper<S>
+where
+    S: Service<Req, Response = Res, Error = Err>,
+    S::Future: 'static,
+{
+    type Response = Res;
+    type Error = Err;
+    type Future = BoxFuture<Result<Res, Err>>;
+
+    crate::forward_ready!(inner);
+
+    fn call(&self, req: Req) -> Self::Future {
+        Box::pin(self.inner.call(req))
+    }
+}
+
+/// Wrapper for a service factory trait object that will produce a boxed trait object service.
 pub struct BoxServiceFactory<Cfg, Req, Res, Err, InitErr>(Inner<Cfg, Req, Res, Err, InitErr>);
 
-/// Create boxed service factory
+/// Create service factory trait object.
 pub fn factory<SF, Req>(
     factory: SF,
 ) -> BoxServiceFactory<SF::Config, Req, SF::Response, SF::Error, SF::InitError>
@@ -28,20 +72,7 @@ where
     SF::Error: 'static,
     SF::InitError: 'static,
 {
-    BoxServiceFactory(Box::new(FactoryWrapper {
+    BoxServiceFactory(Box::new(FactoryWrapper(factory)))
-        factory,
-        _t: PhantomData,
-    }))
-}
-
-/// Create boxed service
-pub fn service<S, Req>(service: S) -> BoxService<Req, S::Response, S::Error>
-where
-    S: Service<Req> + 'static,
-    Req: 'static,
-    S::Future: 'static,
-{
-    Box::new(ServiceWrapper(service, PhantomData))
 }
 
 type Inner<C, Req, Res, Err, InitErr> = Box<
@@ -66,9 +97,9 @@ where
 {
     type Response = Res;
     type Error = Err;
-    type InitError = InitErr;
     type Config = C;
     type Service = BoxService<Req, Res, Err>;
+    type InitError = InitErr;
 
     type Future = BoxFuture<Result<Self::Service, InitErr>>;
 
@@ -77,12 +108,9 @@ where
     }
 }
 
-struct FactoryWrapper<SF, Req, Cfg> {
+struct FactoryWrapper<SF>(SF);
-    factory: SF,
-    _t: PhantomData<(Req, Cfg)>,
-}
 
-impl<SF, Req, Cfg, Res, Err, InitErr> ServiceFactory<Req> for FactoryWrapper<SF, Req, Cfg>
+impl<SF, Req, Cfg, Res, Err, InitErr> ServiceFactory<Req> for FactoryWrapper<SF>
 where
     Req: 'static,
     Res: 'static,
@@ -95,47 +123,13 @@ where
 {
     type Response = Res;
     type Error = Err;
-    type InitError = InitErr;
     type Config = Cfg;
     type Service = BoxService<Req, Res, Err>;
+    type InitError = InitErr;
     type Future = BoxFuture<Result<Self::Service, Self::InitError>>;
 
     fn new_service(&self, cfg: Cfg) -> Self::Future {
-        let fut = self.factory.new_service(cfg);
+        let f = self.0.new_service(cfg);
-        Box::pin(async {
+        Box::pin(async { f.await.map(|s| Box::new(ServiceWrapper::new(s)) as _) })
-            let res = fut.await;
-            res.map(ServiceWrapper::boxed)
-        })
-    }
-}
-
-struct ServiceWrapper<S: Service<Req>, Req>(S, PhantomData<Req>);
-
-impl<S, Req> ServiceWrapper<S, Req>
-where
-    S: Service<Req> + 'static,
-    Req: 'static,
-    S::Future: 'static,
-{
-    fn boxed(service: S) -> BoxService<Req, S::Response, S::Error> {
-        Box::new(ServiceWrapper(service, PhantomData))
-    }
-}
-
-impl<S, Req, Res, Err> Service<Req> for ServiceWrapper<S, Req>
-where
-    S: Service<Req, Response = Res, Error = Err>,
-    S::Future: 'static,
-{
-    type Response = Res;
-    type Error = Err;
-    type Future = BoxFuture<Result<Res, Err>>;
-
-    fn poll_ready(&self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.0.poll_ready(ctx)
-    }
-
-    fn call(&self, req: Req) -> Self::Future {
-        Box::pin(self.0.call(req))
     }
 }

actix-service/src/ext.rs

@@ -1,4 +1,7 @@
-use crate::{dev, Service, ServiceFactory};
+use crate::{
+    map::Map, map_err::MapErr, transform_err::TransformMapInitErr, Service, ServiceFactory,
+    Transform,
+};
 
 pub trait ServiceExt<Req>: Service<Req> {
     /// Map this service's output to a different type, returning a new service
@@ -10,12 +13,12 @@ pub trait ServiceExt<Req>: Service<Req> {
     /// Note that this function consumes the receiving service and returns a
     /// wrapped version of it, similar to the existing `map` methods in the
     /// standard library.
-    fn map<F, R>(self, f: F) -> dev::Map<Self, F, Req, R>
+    fn map<F, R>(self, f: F) -> Map<Self, F, Req, R>
     where
         Self: Sized,
         F: FnMut(Self::Response) -> R,
     {
-        dev::Map::new(self, f)
+        Map::new(self, f)
     }
 
     /// Map this service's error to a different error, returning a new service.
@@ -26,12 +29,12 @@ pub trait ServiceExt<Req>: Service<Req> {
     ///
     /// Note that this function consumes the receiving service and returns a
     /// wrapped version of it.
-    fn map_err<F, E>(self, f: F) -> dev::MapErr<Self, Req, F, E>
+    fn map_err<F, E>(self, f: F) -> MapErr<Self, Req, F, E>
     where
         Self: Sized,
         F: Fn(Self::Error) -> E,
     {
-        dev::MapErr::new(self, f)
+        MapErr::new(self, f)
     }
 }
 
@@ -67,4 +70,17 @@ pub trait ServiceFactoryExt<Req>: ServiceFactory<Req> {
     }
 }
 
-impl<S, Req> ServiceFactoryExt<Req> for S where S: ServiceFactory<Req> {}
+impl<SF, Req> ServiceFactoryExt<Req> for SF where SF: ServiceFactory<Req> {}
+
+pub trait TransformExt<S, Req>: Transform<S, Req> {
+    /// Return a new `Transform` whose init error is mapped to to a different type.
+    fn map_init_err<F, E>(self, f: F) -> TransformMapInitErr<Self, S, Req, F, E>
+    where
+        Self: Sized,
+        F: Fn(Self::InitError) -> E + Clone,
+    {
+        TransformMapInitErr::new(self, f)
+    }
+}
+
+impl<T, Req> TransformExt<T, Req> for T where T: Transform<T, Req> {}

actix-service/src/fn_service.rs

@@ -1,4 +1,4 @@
-use core::{future::Future, marker::PhantomData, task::Poll};
+use core::{future::Future, marker::PhantomData};
 
 use crate::{ok, IntoService, IntoServiceFactory, Ready, Service, ServiceFactory};
 
@@ -15,8 +15,7 @@ where
 
 /// Create `ServiceFactory` for function that can produce services
 ///
-/// # Example
+/// # Examples
-///
 /// ```
 /// use std::io;
 /// use actix_service::{fn_factory, fn_service, Service, ServiceFactory};
@@ -62,11 +61,10 @@ where
 
 /// Create `ServiceFactory` for function that accepts config argument and can produce services
 ///
-/// Any function that has following form `Fn(Config) -> Future<Output = Service>` could
+/// Any function that has following form `Fn(Config) -> Future<Output = Service>` could act as
-/// act as a `ServiceFactory`.
+/// a `ServiceFactory`.
-///
-/// # Example
 ///
+/// # Examples
 /// ```
 /// use std::io;
 /// use actix_service::{fn_factory_with_config, fn_service, Service, ServiceFactory};

actix-service/src/lib.rs

@@ -21,6 +21,7 @@ mod apply_cfg;
 pub mod boxed;
 mod ext;
 mod fn_service;
+mod macros;
 mod map;
 mod map_config;
 mod map_err;
@@ -33,11 +34,11 @@ mod transform_err;
 
 pub use self::apply::{apply_fn, apply_fn_factory};
 pub use self::apply_cfg::{apply_cfg, apply_cfg_factory};
-pub use self::ext::{ServiceExt, ServiceFactoryExt};
+pub use self::ext::{ServiceExt, ServiceFactoryExt, TransformExt};
 pub use self::fn_service::{fn_factory, fn_factory_with_config, fn_service};
 pub use self::map_config::{map_config, unit_config};
 pub use self::pipeline::{pipeline, pipeline_factory, Pipeline, PipelineFactory};
-pub use self::transform::{apply, Transform};
+pub use self::transform::{apply, ApplyTransform, Transform};
 
 #[allow(unused_imports)]
 use self::ready::{err, ok, ready, Ready};
@@ -150,6 +151,7 @@ pub trait ServiceFactory<Req> {
     fn new_service(&self, cfg: Self::Config) -> Self::Future;
 }
 
+// TODO: remove implement on mut reference.
 impl<'a, S, Req> Service<Req> for &'a mut S
 where
     S: Service<Req> + 'a,
@@ -167,6 +169,23 @@ where
     }
 }
 
+impl<'a, S, Req> Service<Req> for &'a S
+where
+    S: Service<Req> + 'a,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    fn poll_ready(&self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        (**self).poll_ready(ctx)
+    }
+
+    fn call(&self, request: Req) -> S::Future {
+        (**self).call(request)
+    }
+}
+
 impl<S, Req> Service<Req> for Box<S>
 where
     S: Service<Req> + ?Sized,
@@ -184,24 +203,25 @@ where
     }
 }
 
-impl<S, Req> Service<Req> for RefCell<S>
+impl<S, Req> Service<Req> for Rc<S>
 where
-    S: Service<Req>,
+    S: Service<Req> + ?Sized,
 {
     type Response = S::Response;
     type Error = S::Error;
     type Future = S::Future;
 
     fn poll_ready(&self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.borrow().poll_ready(ctx)
+        (**self).poll_ready(ctx)
     }
 
     fn call(&self, request: Req) -> S::Future {
-        self.borrow().call(request)
+        (**self).call(request)
     }
 }
 
-impl<S, Req> Service<Req> for Rc<RefCell<S>>
+/// This impl is deprecated since v2 because the `Service` trait now receives shared reference.
+impl<S, Req> Service<Req> for RefCell<S>
 where
     S: Service<Req>,
 {
@@ -294,44 +314,3 @@ where
 {
     tp.into_service()
 }
-
-pub mod dev {
-    pub use crate::apply::{Apply, ApplyFactory};
-    pub use crate::fn_service::{
-        FnService, FnServiceConfig, FnServiceFactory, FnServiceNoConfig,
-    };
-    pub use crate::map::{Map, MapServiceFactory};
-    pub use crate::map_config::{MapConfig, UnitConfig};
-    pub use crate::map_err::{MapErr, MapErrServiceFactory};
-    pub use crate::map_init_err::MapInitErr;
-    pub use crate::transform::ApplyTransform;
-    pub use crate::transform_err::TransformMapInitErr;
-}
-
-#[macro_export]
-macro_rules! always_ready {
-    () => {
-        #[inline]
-        fn poll_ready(
-            &self,
-            _: &mut ::core::task::Context<'_>,
-        ) -> ::core::task::Poll<Result<(), Self::Error>> {
-            Poll::Ready(Ok(()))
-        }
-    };
-}
-
-#[macro_export]
-macro_rules! forward_ready {
-    ($field:ident) => {
-        #[inline]
-        fn poll_ready(
-            &self,
-            cx: &mut ::core::task::Context<'_>,
-        ) -> ::core::task::Poll<Result<(), Self::Error>> {
-            self.$field
-                .poll_ready(cx)
-                .map_err(::core::convert::Into::into)
-        }
-    };
-}

actix-service/src/macros.rs

@@ -0,0 +1,181 @@
+/// A boilerplate implementation of [`Service::poll_ready`] that always signals readiness.
+///
+/// [`Service::poll_ready`]: crate::Service::poll_ready
+///
+/// # Examples
+/// ```no_run
+/// use actix_service::Service;
+/// use futures_util::future::{ready, Ready};
+///
+/// struct IdentityService;
+///
+/// impl Service<u32> for IdentityService {
+///     type Response = u32;
+///     type Error = ();
+///     type Future = Ready<Result<Self::Response, Self::Error>>;
+///
+///     actix_service::always_ready!();
+///
+///     fn call(&self, req: u32) -> Self::Future {
+///         ready(Ok(req))
+///     }
+/// }
+/// ```
+#[macro_export]
+macro_rules! always_ready {
+    () => {
+        #[inline]
+        fn poll_ready(
+            &self,
+            _: &mut ::core::task::Context<'_>,
+        ) -> ::core::task::Poll<Result<(), Self::Error>> {
+            ::core::task::Poll::Ready(Ok(()))
+        }
+    };
+}
+
+/// A boilerplate implementation of [`Service::poll_ready`] that forwards readiness checks to a
+/// named struct field.
+///
+/// Tuple structs are not supported.
+///
+/// [`Service::poll_ready`]: crate::Service::poll_ready
+///
+/// # Examples
+/// ```no_run
+/// use actix_service::Service;
+/// use futures_util::future::{ready, Ready};
+///
+/// struct WrapperService<S> {
+///     inner: S,
+/// }
+///
+/// impl<S> Service<()> for WrapperService<S>
+/// where
+///     S: Service<()>,
+/// {
+///     type Response = S::Response;
+///     type Error = S::Error;
+///     type Future = S::Future;
+///
+///     actix_service::forward_ready!(inner);
+///
+///     fn call(&self, req: ()) -> Self::Future {
+///         self.inner.call(req)
+///     }
+/// }
+/// ```
+#[macro_export]
+macro_rules! forward_ready {
+    ($field:ident) => {
+        #[inline]
+        fn poll_ready(
+            &self,
+            cx: &mut ::core::task::Context<'_>,
+        ) -> ::core::task::Poll<Result<(), Self::Error>> {
+            self.$field
+                .poll_ready(cx)
+                .map_err(::core::convert::Into::into)
+        }
+    };
+}
+
+#[cfg(test)]
+mod tests {
+    use core::{
+        cell::Cell,
+        convert::Infallible,
+        task::{self, Context, Poll},
+    };
+
+    use futures_util::{
+        future::{ready, Ready},
+        task::noop_waker,
+    };
+
+    use crate::Service;
+
+    struct IdentityService;
+
+    impl Service<u32> for IdentityService {
+        type Response = u32;
+        type Error = Infallible;
+        type Future = Ready<Result<Self::Response, Self::Error>>;
+
+        always_ready!();
+
+        fn call(&self, req: u32) -> Self::Future {
+            ready(Ok(req))
+        }
+    }
+
+    struct CountdownService(Cell<u32>);
+
+    impl Service<()> for CountdownService {
+        type Response = ();
+        type Error = Infallible;
+        type Future = Ready<Result<Self::Response, Self::Error>>;
+
+        fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+            let count = self.0.get();
+
+            if count == 0 {
+                Poll::Ready(Ok(()))
+            } else {
+                self.0.set(count - 1);
+                cx.waker().wake_by_ref();
+                Poll::Pending
+            }
+        }
+
+        fn call(&self, _: ()) -> Self::Future {
+            ready(Ok(()))
+        }
+    }
+
+    struct WrapperService<S> {
+        inner: S,
+    }
+
+    impl<S> Service<()> for WrapperService<S>
+    where
+        S: Service<()>,
+    {
+        type Response = S::Response;
+        type Error = S::Error;
+        type Future = S::Future;
+
+        forward_ready!(inner);
+
+        fn call(&self, _: ()) -> Self::Future {
+            self.inner.call(())
+        }
+    }
+
+    #[test]
+    fn test_always_ready_macro() {
+        let waker = noop_waker();
+        let mut cx = task::Context::from_waker(&waker);
+
+        let svc = IdentityService;
+
+        assert!(svc.poll_ready(&mut cx).is_ready());
+        assert!(svc.poll_ready(&mut cx).is_ready());
+        assert!(svc.poll_ready(&mut cx).is_ready());
+    }
+
+    #[test]
+    fn test_forward_ready_macro() {
+        let waker = noop_waker();
+        let mut cx = task::Context::from_waker(&waker);
+
+        let svc = WrapperService {
+            inner: CountdownService(Cell::new(3)),
+        };
+
+        assert!(svc.poll_ready(&mut cx).is_pending());
+        assert!(svc.poll_ready(&mut cx).is_pending());
+        assert!(svc.poll_ready(&mut cx).is_pending());
+        assert!(svc.poll_ready(&mut cx).is_ready());
+    }
+}

actix-service/src/map_err.rs

@@ -180,7 +180,7 @@ where
     F: Fn(A::Error) -> E,
 {
     fn new(fut: A::Future, f: F) -> Self {
-        MapErrServiceFuture { f, fut }
+        MapErrServiceFuture { fut, f }
     }
 }
 

actix-service/src/transform.rs

@@ -9,10 +9,9 @@ use core::{
 use futures_core::ready;
 use pin_project_lite::pin_project;
 
-use crate::transform_err::TransformMapInitErr;
 use crate::{IntoServiceFactory, Service, ServiceFactory};
 
-/// Apply transform to a service.
+/// Apply a [`Transform`] to a [`Service`].
 pub fn apply<T, S, I, Req>(t: T, factory: I) -> ApplyTransform<T, S, Req>
 where
     I: IntoServiceFactory<S, Req>,
@@ -25,9 +24,8 @@ where
 /// The `Transform` trait defines the interface of a service factory that wraps inner service
 /// during construction.
 ///
-/// Transform(middleware) wraps inner service and runs during
+/// Transform(middleware) wraps inner service and runs during inbound and/or outbound processing in
-/// inbound and/or outbound processing in the request/response lifecycle.
+/// the request/response lifecycle. It may modify request and/or response.
-/// It may modify request and/or response.
 ///
 /// For example, timeout transform:
 ///
@@ -51,20 +49,19 @@ where
 ///     fn call(&self, req: S::Request) -> Self::Future {
 ///         TimeoutServiceResponse {
 ///             fut: self.service.call(req),
-///             sleep: Delay::new(clock::now() + self.timeout),
+///             sleep: Sleep::new(clock::now() + self.timeout),
 ///         }
 ///     }
 /// }
 /// ```
 ///
-/// Timeout service in above example is decoupled from underlying service implementation
+/// Timeout service in above example is decoupled from underlying service implementation and could
-/// and could be applied to any service.
+/// be applied to any service.
 ///
-/// The `Transform` trait defines the interface of a Service factory. `Transform`
+/// The `Transform` trait defines the interface of a Service factory. `Transform` is often
-/// is often implemented for middleware, defining how to construct a
+/// implemented for middleware, defining how to construct a middleware Service. A Service that is
-/// middleware Service. A Service that is constructed by the factory takes
+/// constructed by the factory takes the Service that follows it during execution as a parameter,
-/// the Service that follows it during execution as a parameter, assuming
+/// assuming ownership of the next Service.
-/// ownership of the next Service.
 ///
 /// Factory for `Timeout` middleware from the above example could look like this:
 ///
@@ -85,15 +82,15 @@ where
 ///     type Future = Ready<Result<Self::Transform, Self::InitError>>;
 ///
 ///     fn new_transform(&self, service: S) -> Self::Future {
-///         ok(TimeoutService {
+///         ready(Ok(TimeoutService {
 ///             service,
 ///             timeout: self.timeout,
-///         })
+///         }))
 ///     }
 /// }
 /// ```
 pub trait Transform<S, Req> {
-    /// Responses given by the service.
+    /// Responses produced by the service.
     type Response;
 
     /// Errors produced by the service.
@@ -110,16 +107,6 @@ pub trait Transform<S, Req> {
 
     /// Creates and returns a new Transform component, asynchronously
     fn new_transform(&self, service: S) -> Self::Future;
-
-    /// Map this transform's factory error to a different error,
-    /// returning a new transform service factory.
-    fn map_init_err<F, E>(self, f: F) -> TransformMapInitErr<Self, S, Req, F, E>
-    where
-        Self: Sized,
-        F: Fn(Self::InitError) -> E + Clone,
-    {
-        TransformMapInitErr::new(self, f)
-    }
 }
 
 impl<T, S, Req> Transform<S, Req> for Rc<T>
@@ -152,7 +139,7 @@ where
     }
 }
 
-/// `Apply` transform to new service
+/// Apply a [`Transform`] to a [`Service`].
 pub struct ApplyTransform<T, S, Req>(Rc<(T, S)>, PhantomData<Req>);
 
 impl<T, S, Req> ApplyTransform<T, S, Req>

actix-service/src/transform_err.rs

@@ -9,10 +9,8 @@ use pin_project_lite::pin_project;
 
 use super::Transform;
 
-/// Transform for the `map_init_err` combinator, changing the type of a new
+/// Transform for the [`TransformExt::map_init_err`] combinator, changing the type of a new
-/// transform's init error.
+/// [`Transform`]'s initialization error.
-///
-/// This is created by the `Transform::map_init_err` method.
 pub struct TransformMapInitErr<T, S, Req, F, E> {
     transform: T,
     mapper: F,

actix-tls/CHANGES.md

@@ -3,6 +3,22 @@
 ## Unreleased - 2021-xx-xx
 
 
+## 3.0.0-beta.5 - 2021-03-29
+* Changed `connect::ssl::rustls::RustlsConnectorService` to return error when `DNSNameRef` 
+  generation failed instead of panic. [#296]
+* Remove `connect::ssl::openssl::OpensslConnectServiceFactory`. [#297]
+* Remove `connect::ssl::openssl::OpensslConnectService`. [#297]
+* Add `connect::ssl::native_tls` module for native tls support. [#295]
+* Rename `accept::{nativetls => native_tls}`. [#295]
+* Remove `connect::TcpConnectService` type. service caller expect a `TcpStream` should use 
+  `connect::ConnectService` instead and call `Connection<T, TcpStream>::into_parts`. [#299]
+
+[#295]: https://github.com/actix/actix-net/pull/295
+[#296]: https://github.com/actix/actix-net/pull/296
+[#297]: https://github.com/actix/actix-net/pull/297
+[#299]: https://github.com/actix/actix-net/pull/299
+
+
 ## 3.0.0-beta.4 - 2021-02-24
 * Rename `accept::openssl::{SslStream => TlsStream}`.
 * Add `connect::Connect::set_local_addr` to attach local `IpAddr`. [#282]

actix-tls/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-tls"
-version = "3.0.0-beta.4"
+version = "3.0.0-beta.5"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "TLS acceptor and connector services for Actix ecosystem"
 keywords = ["network", "tls", "ssl", "async", "transport"]
@@ -41,8 +41,8 @@ uri = ["http"]
 
 [dependencies]
 actix-codec = "0.4.0-beta.1"
-actix-rt = { version = "2.1.0", default-features = false }
+actix-rt = { version = "2.2.0", default-features = false }
-actix-service = "2.0.0-beta.4"
+actix-service = "2.0.0-beta.5"
 actix-utils = "3.0.0-beta.2"
 
 derive_more = "0.99.5"
@@ -62,14 +62,8 @@ webpki-roots = { version = "0.21", optional = true }
 # native-tls
 tokio-native-tls = { version = "0.3", optional = true }
 
-[target.'cfg(windows)'.dependencies.tls-openssl]
-version = "0.10.9"
-package = "openssl"
-features = ["vendored"]
-optional = true
-
 [dev-dependencies]
-actix-rt = "2.1.0"
+actix-rt = "2.2.0"
 actix-server = "2.0.0-beta.3"
 bytes = "1"
 env_logger = "0.8"

actix-tls/src/accept/mod.rs

@@ -16,7 +16,7 @@ pub mod openssl;
 pub mod rustls;
 
 #[cfg(feature = "native-tls")]
-pub mod nativetls;
+pub mod native_tls;
 
 pub(crate) static MAX_CONN: AtomicUsize = AtomicUsize::new(256);
 

actix-tls/src/accept/native_tls.rs

@@ -6,7 +6,7 @@ use std::{
 };
 
 use actix_codec::{AsyncRead, AsyncWrite, ReadBuf};
-use actix_rt::net::ActixStream;
+use actix_rt::net::{ActixStream, Ready};
 use actix_service::{Service, ServiceFactory};
 use actix_utils::counter::Counter;
 use futures_core::future::LocalBoxFuture;
@@ -80,11 +80,11 @@ impl<T: ActixStream> AsyncWrite for TlsStream<T> {
 }
 
 impl<T: ActixStream> ActixStream for TlsStream<T> {
-    fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+    fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
         T::poll_read_ready((&**self).get_ref().get_ref().get_ref(), cx)
     }
 
-    fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+    fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
         T::poll_write_ready((&**self).get_ref().get_ref().get_ref(), cx)
     }
 }
@@ -113,7 +113,7 @@ impl Clone for Acceptor {
     }
 }
 
-impl<T: ActixStream> ServiceFactory<T> for Acceptor {
+impl<T: ActixStream + 'static> ServiceFactory<T> for Acceptor {
     type Response = TlsStream<T>;
     type Error = Error;
     type Config = ();
@@ -138,16 +138,7 @@ pub struct NativeTlsAcceptorService {
     conns: Counter,
 }
 
-impl Clone for NativeTlsAcceptorService {
+impl<T: ActixStream + 'static> Service<T> for NativeTlsAcceptorService {
-    fn clone(&self) -> Self {
-        Self {
-            acceptor: self.acceptor.clone(),
-            conns: self.conns.clone(),
-        }
-    }
-}
-
-impl<T: ActixStream> Service<T> for NativeTlsAcceptorService {
     type Response = TlsStream<T>;
     type Error = Error;
     type Future = LocalBoxFuture<'static, Result<TlsStream<T>, Error>>;
@@ -162,9 +153,9 @@ impl<T: ActixStream> Service<T> for NativeTlsAcceptorService {
 
     fn call(&self, io: T) -> Self::Future {
         let guard = self.conns.get();
-        let this = self.clone();
+        let acceptor = self.acceptor.clone();
         Box::pin(async move {
-            let io = this.acceptor.accept(io).await;
+            let io = acceptor.accept(io).await;
             drop(guard);
             io.map(Into::into)
         })

actix-tls/src/accept/openssl.rs

@@ -7,7 +7,7 @@ use std::{
 };
 
 use actix_codec::{AsyncRead, AsyncWrite, ReadBuf};
-use actix_rt::net::ActixStream;
+use actix_rt::net::{ActixStream, Ready};
 use actix_service::{Service, ServiceFactory};
 use actix_utils::counter::{Counter, CounterGuard};
 use futures_core::{future::LocalBoxFuture, ready};
@@ -82,11 +82,11 @@ impl<T: ActixStream> AsyncWrite for TlsStream<T> {
 }
 
 impl<T: ActixStream> ActixStream for TlsStream<T> {
-    fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+    fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
         T::poll_read_ready((&**self).get_ref(), cx)
     }
 
-    fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+    fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
         T::poll_write_ready((&**self).get_ref(), cx)
     }
 }

actix-tls/src/accept/rustls.rs

@@ -8,7 +8,7 @@ use std::{
 };
 
 use actix_codec::{AsyncRead, AsyncWrite, ReadBuf};
-use actix_rt::net::ActixStream;
+use actix_rt::net::{ActixStream, Ready};
 use actix_service::{Service, ServiceFactory};
 use actix_utils::counter::{Counter, CounterGuard};
 use futures_core::future::LocalBoxFuture;
@@ -82,11 +82,11 @@ impl<T: ActixStream> AsyncWrite for TlsStream<T> {
 }
 
 impl<T: ActixStream> ActixStream for TlsStream<T> {
-    fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+    fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
         T::poll_read_ready((&**self).get_ref().0, cx)
     }
 
-    fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+    fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<Ready>> {
         T::poll_write_ready((&**self).get_ref().0, cx)
     }
 }

actix-tls/src/connect/connector.rs

@@ -72,7 +72,7 @@ pub enum TcpConnectorResponse<T> {
         port: u16,
         local_addr: Option<IpAddr>,
         addrs: Option<VecDeque<SocketAddr>>,
-        stream: Option<ReusableBoxFuture<Result<TcpStream, io::Error>>>,
+        stream: ReusableBoxFuture<Result<TcpStream, io::Error>>,
     },
     Error(Option<ConnectError>),
 }
@@ -103,18 +103,22 @@ impl<T: Address> TcpConnectorResponse<T> {
                 port,
                 local_addr,
                 addrs: None,
-                stream: Some(ReusableBoxFuture::new(connect(addr, local_addr))),
+                stream: ReusableBoxFuture::new(connect(addr, local_addr)),
             },
 
             // when resolver returns multiple socket addr for request they would be popped from
             // front end of queue and returns with the first successful tcp connection.
-            ConnectAddrs::Multi(addrs) => TcpConnectorResponse::Response {
+            ConnectAddrs::Multi(mut addrs) => {
-                req: Some(req),
+                let addr = addrs.pop_front().unwrap();
-                port,
+
-                local_addr,
+                TcpConnectorResponse::Response {
-                addrs: Some(addrs),
+                    req: Some(req),
-                stream: None,
+                    port,
-            },
+                    local_addr,
+                    addrs: Some(addrs),
+                    stream: ReusableBoxFuture::new(connect(addr, local_addr)),
+                }
+            }
         }
     }
 }
@@ -133,40 +137,31 @@ impl<T: Address> Future for TcpConnectorResponse<T> {
                 addrs,
                 stream,
             } => loop {
-                if let Some(new) = stream.as_mut() {
+                match ready!(stream.poll(cx)) {
-                    match ready!(new.poll(cx)) {
+                    Ok(sock) => {
-                        Ok(sock) => {
+                        let req = req.take().unwrap();
-                            let req = req.take().unwrap();
+                        trace!(
-                            trace!(
+                            "TCP connector: successfully connected to {:?} - {:?}",
-                                "TCP connector: successfully connected to {:?} - {:?}",
+                            req.hostname(),
-                                req.hostname(),
+                            sock.peer_addr()
-                                sock.peer_addr()
+                        );
-                            );
+                        return Poll::Ready(Ok(Connection::new(sock, req)));
-                            return Poll::Ready(Ok(Connection::new(sock, req)));
+                    }
-                        }
 
-                        Err(err) => {
+                    Err(err) => {
-                            trace!(
+                        trace!(
-                                "TCP connector: failed to connect to {:?} port: {}",
+                            "TCP connector: failed to connect to {:?} port: {}",
-                                req.as_ref().unwrap().hostname(),
+                            req.as_ref().unwrap().hostname(),
-                                port,
+                            port,
-                            );
+                        );
 
-                            if addrs.is_none() || addrs.as_ref().unwrap().is_empty() {
+                        if let Some(addr) = addrs.as_mut().and_then(|addrs| addrs.pop_front()) {
-                                return Poll::Ready(Err(ConnectError::Io(err)));
+                            stream.set(connect(addr, *local_addr));
-                            }
+                        } else {
+                            return Poll::Ready(Err(ConnectError::Io(err)));
                         }
                     }
                 }
-
-                // try to connect
-                let addr = addrs.as_mut().unwrap().pop_front().unwrap();
-
-                let fut = connect(addr, *local_addr);
-                match stream {
-                    Some(rbf) => rbf.set(fut),
-                    None => *stream = Some(ReusableBoxFuture::new(fut)),
-                }
             },
         }
     }

actix-tls/src/connect/mod.rs

@@ -26,20 +26,20 @@ pub mod ssl;
 mod uri;
 
 use actix_rt::net::TcpStream;
-use actix_service::{pipeline, pipeline_factory, Service, ServiceFactory};
+use actix_service::{Service, ServiceFactory};
 
 pub use self::connect::{Address, Connect, Connection};
 pub use self::connector::{TcpConnector, TcpConnectorFactory};
 pub use self::error::ConnectError;
 pub use self::resolve::{Resolve, Resolver, ResolverFactory};
-pub use self::service::{ConnectService, ConnectServiceFactory, TcpConnectService};
+pub use self::service::{ConnectService, ConnectServiceFactory};
 
 /// Create TCP connector service.
 pub fn new_connector<T: Address + 'static>(
     resolver: Resolver,
 ) -> impl Service<Connect<T>, Response = Connection<T, TcpStream>, Error = ConnectError> + Clone
 {
-    pipeline(resolver).and_then(TcpConnector)
+    ConnectServiceFactory::new(resolver).service()
 }
 
 /// Create TCP connector service factory.
@@ -52,7 +52,7 @@ pub fn new_connector_factory<T: Address + 'static>(
     Error = ConnectError,
     InitError = (),
 > + Clone {
-    pipeline_factory(ResolverFactory::new(resolver)).and_then(TcpConnectorFactory)
+    ConnectServiceFactory::new(resolver)
 }
 
 /// Create connector service with default parameters.

actix-tls/src/connect/service.rs

@@ -34,14 +34,6 @@ impl ConnectServiceFactory {
             resolver: self.resolver.service(),
         }
     }
-
-    /// Construct new tcp stream service
-    pub fn tcp_service(&self) -> TcpConnectService {
-        TcpConnectService {
-            tcp: self.tcp.service(),
-            resolver: self.resolver.service(),
-        }
-    }
 }
 
 impl Clone for ConnectServiceFactory {
@@ -63,7 +55,7 @@ impl<T: Address> ServiceFactory<Connect<T>> for ConnectServiceFactory {
 
     fn new_service(&self, _: ()) -> Self::Future {
         let service = self.service();
-        Box::pin(async move { Ok(service) })
+        Box::pin(async { Ok(service) })
     }
 }
 
@@ -135,44 +127,3 @@ impl<T: Address> Future for ConnectServiceResponse<T> {
         }
     }
 }
-
-#[derive(Clone)]
-pub struct TcpConnectService {
-    tcp: TcpConnector,
-    resolver: Resolver,
-}
-
-impl<T: Address> Service<Connect<T>> for TcpConnectService {
-    type Response = TcpStream;
-    type Error = ConnectError;
-    type Future = TcpConnectServiceResponse<T>;
-
-    actix_service::always_ready!();
-
-    fn call(&self, req: Connect<T>) -> Self::Future {
-        TcpConnectServiceResponse {
-            fut: ConnectFuture::Resolve(self.resolver.call(req)),
-            tcp: self.tcp,
-        }
-    }
-}
-
-pub struct TcpConnectServiceResponse<T: Address> {
-    fut: ConnectFuture<T>,
-    tcp: TcpConnector,
-}
-
-impl<T: Address> Future for TcpConnectServiceResponse<T> {
-    type Output = Result<TcpStream, ConnectError>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        loop {
-            match ready!(self.fut.poll_connect(cx))? {
-                ConnectOutput::Resolved(res) => {
-                    self.fut = ConnectFuture::Connect(self.tcp.call(res));
-                }
-                ConnectOutput::Connected(conn) => return Poll::Ready(Ok(conn.into_parts().0)),
-            }
-        }
-    }
-}

actix-tls/src/connect/ssl/mod.rs

@@ -5,3 +5,6 @@ pub mod openssl;
 
 #[cfg(feature = "rustls")]
 pub mod rustls;
+
+#[cfg(feature = "native-tls")]
+pub mod native_tls;

actix-tls/src/connect/ssl/native_tls.rs

@@ -0,0 +1,88 @@
+use std::io;
+
+use actix_rt::net::ActixStream;
+use actix_service::{Service, ServiceFactory};
+use futures_core::future::LocalBoxFuture;
+use log::trace;
+use tokio_native_tls::{TlsConnector as TokioNativetlsConnector, TlsStream};
+
+pub use tokio_native_tls::native_tls::TlsConnector;
+
+use crate::connect::{Address, Connection};
+
+/// Native-tls connector factory and service
+pub struct NativetlsConnector {
+    connector: TokioNativetlsConnector,
+}
+
+impl NativetlsConnector {
+    pub fn new(connector: TlsConnector) -> Self {
+        Self {
+            connector: TokioNativetlsConnector::from(connector),
+        }
+    }
+}
+
+impl NativetlsConnector {
+    pub fn service(connector: TlsConnector) -> Self {
+        Self::new(connector)
+    }
+}
+
+impl Clone for NativetlsConnector {
+    fn clone(&self) -> Self {
+        Self {
+            connector: self.connector.clone(),
+        }
+    }
+}
+
+impl<T: Address, U> ServiceFactory<Connection<T, U>> for NativetlsConnector
+where
+    U: ActixStream + 'static,
+{
+    type Response = Connection<T, TlsStream<U>>;
+    type Error = io::Error;
+    type Config = ();
+    type Service = Self;
+    type InitError = ();
+    type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;
+
+    fn new_service(&self, _: ()) -> Self::Future {
+        let connector = self.clone();
+        Box::pin(async { Ok(connector) })
+    }
+}
+
+// NativetlsConnector is both it's ServiceFactory and Service impl type.
+// As the factory and service share the same type and state.
+impl<T, U> Service<Connection<T, U>> for NativetlsConnector
+where
+    T: Address,
+    U: ActixStream + 'static,
+{
+    type Response = Connection<T, TlsStream<U>>;
+    type Error = io::Error;
+    type Future = LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;
+
+    actix_service::always_ready!();
+
+    fn call(&self, stream: Connection<T, U>) -> Self::Future {
+        let (io, stream) = stream.replace_io(());
+        let connector = self.connector.clone();
+        Box::pin(async move {
+            trace!("SSL Handshake start for: {:?}", stream.host());
+            connector
+                .connect(stream.host(), io)
+                .await
+                .map(|res| {
+                    trace!("SSL Handshake success: {:?}", stream.host());
+                    stream.replace_io(res).1
+                })
+                .map_err(|e| {
+                    trace!("SSL Handshake error: {:?}", e);
+                    io::Error::new(io::ErrorKind::Other, format!("{}", e))
+                })
+        })
+    }
+}

actix-tls/src/connect/ssl/openssl.rs

@@ -1,13 +1,11 @@
 use std::{
-    fmt,
     future::Future,
     io,
     pin::Pin,
     task::{Context, Poll},
 };
 
-use actix_codec::{AsyncRead, AsyncWrite};
+use actix_rt::net::ActixStream;
-use actix_rt::net::TcpStream;
 use actix_service::{Service, ServiceFactory};
 use futures_core::{future::LocalBoxFuture, ready};
 use log::trace;
@@ -15,10 +13,7 @@ use log::trace;
 pub use openssl::ssl::{Error as SslError, HandshakeError, SslConnector, SslMethod};
 pub use tokio_openssl::SslStream;
 
-use crate::connect::resolve::Resolve;
+use crate::connect::{Address, Connection};
-use crate::connect::{
-    Address, Connect, ConnectError, ConnectService, ConnectServiceFactory, Connection, Resolver,
-};
 
 /// OpenSSL connector factory
 pub struct OpensslConnector {
@@ -45,8 +40,8 @@ impl Clone for OpensslConnector {
 
 impl<T, U> ServiceFactory<Connection<T, U>> for OpensslConnector
 where
-    T: Address + 'static,
+    T: Address,
-    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
+    U: ActixStream + 'static,
 {
     type Response = Connection<T, SslStream<U>>;
     type Error = io::Error;
@@ -75,8 +70,8 @@ impl Clone for OpensslConnectorService {
 
 impl<T, U> Service<Connection<T, U>> for OpensslConnectorService
 where
-    T: Address + 'static,
+    T: Address,
-    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
+    U: ActixStream,
 {
     type Response = Connection<T, SslStream<U>>;
     type Error = io::Error;
@@ -112,7 +107,8 @@ pub struct ConnectAsyncExt<T, U> {
 
 impl<T: Address, U> Future for ConnectAsyncExt<T, U>
 where
-    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug + 'static,
+    T: Address,
+    U: ActixStream,
 {
     type Output = Result<Connection<T, SslStream<U>>, io::Error>;
 
@@ -132,115 +128,3 @@ where
         }
     }
 }
-
-pub struct OpensslConnectServiceFactory {
-    tcp: ConnectServiceFactory,
-    openssl: OpensslConnector,
-}
-
-impl OpensslConnectServiceFactory {
-    /// Construct new OpensslConnectService factory
-    pub fn new(connector: SslConnector) -> Self {
-        OpensslConnectServiceFactory {
-            tcp: ConnectServiceFactory::new(Resolver::Default),
-            openssl: OpensslConnector::new(connector),
-        }
-    }
-
-    /// Construct new connect service with custom DNS resolver
-    pub fn with_resolver(connector: SslConnector, resolver: impl Resolve + 'static) -> Self {
-        OpensslConnectServiceFactory {
-            tcp: ConnectServiceFactory::new(Resolver::new_custom(resolver)),
-            openssl: OpensslConnector::new(connector),
-        }
-    }
-
-    /// Construct OpenSSL connect service
-    pub fn service(&self) -> OpensslConnectService {
-        OpensslConnectService {
-            tcp: self.tcp.service(),
-            openssl: OpensslConnectorService {
-                connector: self.openssl.connector.clone(),
-            },
-        }
-    }
-}
-
-impl Clone for OpensslConnectServiceFactory {
-    fn clone(&self) -> Self {
-        OpensslConnectServiceFactory {
-            tcp: self.tcp.clone(),
-            openssl: self.openssl.clone(),
-        }
-    }
-}
-
-impl<T: Address + 'static> ServiceFactory<Connect<T>> for OpensslConnectServiceFactory {
-    type Response = SslStream<TcpStream>;
-    type Error = ConnectError;
-    type Config = ();
-    type Service = OpensslConnectService;
-    type InitError = ();
-    type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;
-
-    fn new_service(&self, _: ()) -> Self::Future {
-        let service = self.service();
-        Box::pin(async { Ok(service) })
-    }
-}
-
-#[derive(Clone)]
-pub struct OpensslConnectService {
-    tcp: ConnectService,
-    openssl: OpensslConnectorService,
-}
-
-impl<T: Address + 'static> Service<Connect<T>> for OpensslConnectService {
-    type Response = SslStream<TcpStream>;
-    type Error = ConnectError;
-    type Future = OpensslConnectServiceResponse<T>;
-
-    actix_service::always_ready!();
-
-    fn call(&self, req: Connect<T>) -> Self::Future {
-        OpensslConnectServiceResponse {
-            fut1: Some(self.tcp.call(req)),
-            fut2: None,
-            openssl: self.openssl.clone(),
-        }
-    }
-}
-
-pub struct OpensslConnectServiceResponse<T: Address + 'static> {
-    fut1: Option<<ConnectService as Service<Connect<T>>>::Future>,
-    fut2: Option<<OpensslConnectorService as Service<Connection<T, TcpStream>>>::Future>,
-    openssl: OpensslConnectorService,
-}
-
-impl<T: Address> Future for OpensslConnectServiceResponse<T> {
-    type Output = Result<SslStream<TcpStream>, ConnectError>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        if let Some(ref mut fut) = self.fut1 {
-            match ready!(Pin::new(fut).poll(cx)) {
-                Ok(res) => {
-                    let _ = self.fut1.take();
-                    self.fut2 = Some(self.openssl.call(res));
-                }
-                Err(e) => return Poll::Ready(Err(e)),
-            }
-        }
-
-        if let Some(ref mut fut) = self.fut2 {
-            match ready!(Pin::new(fut).poll(cx)) {
-                Ok(connect) => Poll::Ready(Ok(connect.into_parts().0)),
-                Err(e) => Poll::Ready(Err(ConnectError::Io(io::Error::new(
-                    io::ErrorKind::Other,
-                    e,
-                )))),
-            }
-        } else {
-            Poll::Pending
-        }
-    }
-}

actix-tls/src/connect/ssl/rustls.rs

@@ -1,6 +1,6 @@
 use std::{
-    fmt,
     future::Future,
+    io,
     pin::Pin,
     sync::Arc,
     task::{Context, Poll},
@@ -10,7 +10,7 @@ pub use tokio_rustls::rustls::Session;
 pub use tokio_rustls::{client::TlsStream, rustls::ClientConfig};
 pub use webpki_roots::TLS_SERVER_ROOTS;
 
-use actix_codec::{AsyncRead, AsyncWrite};
+use actix_rt::net::ActixStream;
 use actix_service::{Service, ServiceFactory};
 use futures_core::{future::LocalBoxFuture, ready};
 use log::trace;
@@ -44,12 +44,13 @@ impl Clone for RustlsConnector {
     }
 }
 
-impl<T: Address, U> ServiceFactory<Connection<T, U>> for RustlsConnector
+impl<T, U> ServiceFactory<Connection<T, U>> for RustlsConnector
 where
-    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug,
+    T: Address,
+    U: ActixStream + 'static,
 {
     type Response = Connection<T, TlsStream<U>>;
-    type Error = std::io::Error;
+    type Error = io::Error;
     type Config = ();
     type Service = RustlsConnectorService;
     type InitError = ();
@@ -76,43 +77,55 @@ impl Clone for RustlsConnectorService {
 impl<T, U> Service<Connection<T, U>> for RustlsConnectorService
 where
     T: Address,
-    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug,
+    U: ActixStream,
 {
     type Response = Connection<T, TlsStream<U>>;
-    type Error = std::io::Error;
+    type Error = io::Error;
-    type Future = ConnectAsyncExt<T, U>;
+    type Future = RustlsConnectorServiceFuture<T, U>;
 
     actix_service::always_ready!();
 
-    fn call(&self, stream: Connection<T, U>) -> Self::Future {
+    fn call(&self, connection: Connection<T, U>) -> Self::Future {
-        trace!("SSL Handshake start for: {:?}", stream.host());
+        trace!("SSL Handshake start for: {:?}", connection.host());
-        let (io, stream) = stream.replace_io(());
+        let (stream, connection) = connection.replace_io(());
-        let host = DNSNameRef::try_from_ascii_str(stream.host())
+
-            .expect("rustls currently only handles hostname-based connections. See https://github.com/briansmith/webpki/issues/54");
+        match DNSNameRef::try_from_ascii_str(connection.host()) {
-        ConnectAsyncExt {
+            Ok(host) => RustlsConnectorServiceFuture::Future {
-            fut: TlsConnector::from(self.connector.clone()).connect(host, io),
+                connect: TlsConnector::from(self.connector.clone()).connect(host, stream),
-            stream: Some(stream),
+                connection: Some(connection),
+            },
+            Err(_) => RustlsConnectorServiceFuture::InvalidDns,
         }
     }
 }
 
-pub struct ConnectAsyncExt<T, U> {
+pub enum RustlsConnectorServiceFuture<T, U> {
-    fut: Connect<U>,
+    /// See issue https://github.com/briansmith/webpki/issues/54
-    stream: Option<Connection<T, ()>>,
+    InvalidDns,
+    Future {
+        connect: Connect<U>,
+        connection: Option<Connection<T, ()>>,
+    },
 }
 
-impl<T, U> Future for ConnectAsyncExt<T, U>
+impl<T, U> Future for RustlsConnectorServiceFuture<T, U>
 where
     T: Address,
-    U: AsyncRead + AsyncWrite + Unpin + fmt::Debug,
+    U: ActixStream,
 {
-    type Output = Result<Connection<T, TlsStream<U>>, std::io::Error>;
+    type Output = Result<Connection<T, TlsStream<U>>, io::Error>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let this = self.get_mut();
+        match self.get_mut() {
-        let stream = ready!(Pin::new(&mut this.fut).poll(cx))?;
+            Self::InvalidDns => Poll::Ready(Err(
-        let s = this.stream.take().unwrap();
+                io::Error::new(io::ErrorKind::Other, "rustls currently only handles hostname-based connections. See https://github.com/briansmith/webpki/issues/54")
-        trace!("SSL Handshake success: {:?}", s.host());
+            )),
-        Poll::Ready(Ok(s.replace_io(stream).1))
+            Self::Future { connect, connection } => {
+                let stream = ready!(Pin::new(connect).poll(cx))?;
+                let connection = connection.take().unwrap();
+                trace!("SSL Handshake success: {:?}", connection.host());
+                Poll::Ready(Ok(connection.replace_io(stream).1))
+            }
+        }
     }
 }

actix-tracing/Cargo.toml

@@ -16,9 +16,9 @@ name = "actix_tracing"
 path = "src/lib.rs"
 
 [dependencies]
-actix-service = "2.0.0-beta.4"
+actix-service = "2.0.0-beta.5"
 
-futures-util = { version = "0.3.4", default-features = false }
+futures-util = { version = "0.3.7", default-features = false }
 tracing = "0.1"
 tracing-futures = "0.2"
 

actix-tracing/src/lib.rs

@@ -7,7 +7,7 @@
 use core::marker::PhantomData;
 
 use actix_service::{
-    apply, dev::ApplyTransform, IntoServiceFactory, Service, ServiceFactory, Transform,
+    apply, ApplyTransform, IntoServiceFactory, Service, ServiceFactory, Transform,
 };
 use futures_util::future::{ok, Either, Ready};
 use tracing_futures::{Instrument, Instrumented};

actix-utils/CHANGES.md

@@ -3,6 +3,16 @@
 ## Unreleased - 2021-xx-xx
 
 
+## 3.0.0-beta.3 - 2021-04-01
+* Moved `mpsc` to own crate `local-channel`. [#301]
+* Moved `task::LocalWaker` to own crate `local-waker`. [#301]
+* Remove `timeout` module. [#301]
+* Remove `dispatcher` module. [#301]
+* Expose `future` mod with `ready` and `poll_fn` helpers. [#301]
+
+[#301]: https://github.com/actix/actix-net/pull/301
+
+
 ## 3.0.0-beta.2 - 2021-02-06
 * Update `actix-rt` to `2.0.0`. [#273]
 

actix-utils/Cargo.toml

@@ -1,12 +1,13 @@
 [package]
 name = "actix-utils"
-version = "3.0.0-beta.2"
+version = "3.0.0-beta.3"
-authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
+authors = [
-description = "Various network related services and utilities for the Actix ecosystem"
+    "Nikolay Kim <fafhrd91@gmail.com>",
+    "Rob Ede <robjtede@icloud.com>",
+]
+description = "Utilities for the Actix ecosystem"
 keywords = ["network", "framework", "async", "futures"]
-homepage = "https://actix.rs"
 repository = "https://github.com/actix/actix-net.git"
-documentation = "https://docs.rs/actix-utils"
 categories = ["network-programming", "asynchronous"]
 license = "MIT OR Apache-2.0"
 edition = "2018"
@@ -16,14 +17,7 @@ name = "actix_utils"
 path = "src/lib.rs"
 
 [dependencies]
-actix-codec = "0.4.0-beta.1"
+local-waker = "0.1"
-actix-rt = { version = "2.0.0", default-features = false }
-actix-service = "2.0.0-beta.4"
-
-futures-core = { version = "0.3.7", default-features = false }
-futures-sink = { version = "0.3.7", default-features = false }
-log = "0.4"
-pin-project-lite = "0.2.0"
 
 [dev-dependencies]
 actix-rt = "2.0.0"

actix-utils/src/counter.rs

@@ -1,24 +1,18 @@
-use core::cell::Cell;
+//! Task-notifying counter.
-use core::task;
 
+use core::{cell::Cell, fmt, task};
 use std::rc::Rc;
 
-use crate::task::LocalWaker;
+use local_waker::LocalWaker;
 
-#[derive(Clone)]
 /// Simple counter with ability to notify task on reaching specific number
 ///
 /// Counter could be cloned, total n-count is shared across all clones.
+#[derive(Debug, Clone)]
 pub struct Counter(Rc<CounterInner>);
 
-struct CounterInner {
-    count: Cell<usize>,
-    capacity: usize,
-    task: LocalWaker,
-}
-
 impl Counter {
-    /// Create `Counter` instance and set max value.
+    /// Create `Counter` instance with max value.
     pub fn new(capacity: usize) -> Self {
         Counter(Rc::new(CounterInner {
             capacity,
@@ -27,38 +21,26 @@ impl Counter {
         }))
     }
 
-    /// Get counter guard.
+    /// Create new counter guard, incrementing the counter.
     pub fn get(&self) -> CounterGuard {
         CounterGuard::new(self.0.clone())
     }
 
-    /// Check if counter is not at capacity. If counter at capacity
+    /// Notify current task and return true if counter is at capacity.
-    /// it registers notification for current task.
     pub fn available(&self, cx: &mut task::Context<'_>) -> bool {
         self.0.available(cx)
     }
 
-    /// Get total number of acquired counts
+    /// Get total number of acquired guards.
     pub fn total(&self) -> usize {
         self.0.count.get()
     }
 }
 
-pub struct CounterGuard(Rc<CounterInner>);
+struct CounterInner {
-
+    count: Cell<usize>,
-impl CounterGuard {
+    capacity: usize,
-    fn new(inner: Rc<CounterInner>) -> Self {
+    task: LocalWaker,
-        inner.inc();
-        CounterGuard(inner)
-    }
-}
-
-impl Unpin for CounterGuard {}
-
-impl Drop for CounterGuard {
-    fn drop(&mut self) {
-        self.0.dec();
-    }
 }
 
 impl CounterInner {
@@ -83,3 +65,32 @@ impl CounterInner {
         }
     }
 }
+
+impl fmt::Debug for CounterInner {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Counter")
+            .field("count", &self.count.get())
+            .field("capacity", &self.capacity)
+            .field("task", &self.task)
+            .finish()
+    }
+}
+
+/// An RAII structure that keeps the underlying counter incremented until this guard is dropped.
+#[derive(Debug)]
+pub struct CounterGuard(Rc<CounterInner>);
+
+impl CounterGuard {
+    fn new(inner: Rc<CounterInner>) -> Self {
+        inner.inc();
+        CounterGuard(inner)
+    }
+}
+
+impl Unpin for CounterGuard {}
+
+impl Drop for CounterGuard {
+    fn drop(&mut self) {
+        self.0.dec();
+    }
+}

actix-utils/src/dispatcher.rs

@@ -1,338 +0,0 @@
-//! Framed dispatcher service and related utilities.
-
-#![allow(type_alias_bounds)]
-
-use core::future::Future;
-use core::pin::Pin;
-use core::task::{Context, Poll};
-use core::{fmt, mem};
-
-use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
-use actix_service::{IntoService, Service};
-use futures_core::stream::Stream;
-use log::debug;
-use pin_project_lite::pin_project;
-
-use crate::mpsc;
-
-/// Framed transport errors
-pub enum DispatcherError<E, U: Encoder<I> + Decoder, I> {
-    Service(E),
-    Encoder(<U as Encoder<I>>::Error),
-    Decoder(<U as Decoder>::Error),
-}
-
-impl<E, U: Encoder<I> + Decoder, I> From<E> for DispatcherError<E, U, I> {
-    fn from(err: E) -> Self {
-        DispatcherError::Service(err)
-    }
-}
-
-impl<E, U: Encoder<I> + Decoder, I> fmt::Debug for DispatcherError<E, U, I>
-where
-    E: fmt::Debug,
-    <U as Encoder<I>>::Error: fmt::Debug,
-    <U as Decoder>::Error: fmt::Debug,
-{
-    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match *self {
-            DispatcherError::Service(ref e) => write!(fmt, "DispatcherError::Service({:?})", e),
-            DispatcherError::Encoder(ref e) => write!(fmt, "DispatcherError::Encoder({:?})", e),
-            DispatcherError::Decoder(ref e) => write!(fmt, "DispatcherError::Decoder({:?})", e),
-        }
-    }
-}
-
-impl<E, U: Encoder<I> + Decoder, I> fmt::Display for DispatcherError<E, U, I>
-where
-    E: fmt::Display,
-    <U as Encoder<I>>::Error: fmt::Debug,
-    <U as Decoder>::Error: fmt::Debug,
-{
-    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match *self {
-            DispatcherError::Service(ref e) => write!(fmt, "{}", e),
-            DispatcherError::Encoder(ref e) => write!(fmt, "{:?}", e),
-            DispatcherError::Decoder(ref e) => write!(fmt, "{:?}", e),
-        }
-    }
-}
-
-pub enum Message<T> {
-    Item(T),
-    Close,
-}
-
-pin_project! {
-    /// Dispatcher is a future that reads frames from Framed object
-    /// and passes them to the service.
-    pub struct Dispatcher<S, T, U, I>
-    where
-        S: Service<<U as Decoder>::Item, Response = I>,
-        S::Error: 'static,
-        S::Future: 'static,
-        T: AsyncRead,
-        T: AsyncWrite,
-        U: Encoder<I>,
-        U: Decoder,
-        I: 'static,
-        <U as Encoder<I>>::Error: fmt::Debug,
-    {
-        service: S,
-        state: State<S, U, I>,
-        #[pin]
-        framed: Framed<T, U>,
-        rx: mpsc::Receiver<Result<Message<I>, S::Error>>,
-        tx: mpsc::Sender<Result<Message<I>, S::Error>>,
-    }
-}
-
-enum State<S, U, I>
-where
-    S: Service<<U as Decoder>::Item>,
-    U: Encoder<I> + Decoder,
-{
-    Processing,
-    Error(DispatcherError<S::Error, U, I>),
-    FramedError(DispatcherError<S::Error, U, I>),
-    FlushAndStop,
-    Stopping,
-}
-
-impl<S, U, I> State<S, U, I>
-where
-    S: Service<<U as Decoder>::Item>,
-    U: Encoder<I> + Decoder,
-{
-    fn take_error(&mut self) -> DispatcherError<S::Error, U, I> {
-        match mem::replace(self, State::Processing) {
-            State::Error(err) => err,
-            _ => panic!(),
-        }
-    }
-
-    fn take_framed_error(&mut self) -> DispatcherError<S::Error, U, I> {
-        match mem::replace(self, State::Processing) {
-            State::FramedError(err) => err,
-            _ => panic!(),
-        }
-    }
-}
-
-impl<S, T, U, I> Dispatcher<S, T, U, I>
-where
-    S: Service<<U as Decoder>::Item, Response = I>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder<I>,
-    I: 'static,
-    <U as Decoder>::Error: fmt::Debug,
-    <U as Encoder<I>>::Error: fmt::Debug,
-{
-    pub fn new<F>(framed: Framed<T, U>, service: F) -> Self
-    where
-        F: IntoService<S, <U as Decoder>::Item>,
-    {
-        let (tx, rx) = mpsc::channel();
-        Dispatcher {
-            framed,
-            rx,
-            tx,
-            service: service.into_service(),
-            state: State::Processing,
-        }
-    }
-
-    /// Construct new `Dispatcher` instance with customer `mpsc::Receiver`
-    pub fn with_rx<F>(
-        framed: Framed<T, U>,
-        service: F,
-        rx: mpsc::Receiver<Result<Message<I>, S::Error>>,
-    ) -> Self
-    where
-        F: IntoService<S, <U as Decoder>::Item>,
-    {
-        let tx = rx.sender();
-        Dispatcher {
-            framed,
-            rx,
-            tx,
-            service: service.into_service(),
-            state: State::Processing,
-        }
-    }
-
-    /// Get sink
-    pub fn get_sink(&self) -> mpsc::Sender<Result<Message<I>, S::Error>> {
-        self.tx.clone()
-    }
-
-    /// Get reference to a service wrapped by `Dispatcher` instance.
-    pub fn get_ref(&self) -> &S {
-        &self.service
-    }
-
-    /// Get mutable reference to a service wrapped by `Dispatcher` instance.
-    pub fn get_mut(&mut self) -> &mut S {
-        &mut self.service
-    }
-
-    /// Get reference to a framed instance wrapped by `Dispatcher`
-    /// instance.
-    pub fn get_framed(&self) -> &Framed<T, U> {
-        &self.framed
-    }
-
-    /// Get mutable reference to a framed instance wrapped by `Dispatcher` instance.
-    pub fn get_framed_mut(&mut self) -> &mut Framed<T, U> {
-        &mut self.framed
-    }
-
-    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
-    where
-        S: Service<<U as Decoder>::Item, Response = I>,
-        S::Error: 'static,
-        S::Future: 'static,
-        T: AsyncRead + AsyncWrite,
-        U: Decoder + Encoder<I>,
-        I: 'static,
-        <U as Encoder<I>>::Error: fmt::Debug,
-    {
-        loop {
-            let this = self.as_mut().project();
-            match this.service.poll_ready(cx) {
-                Poll::Ready(Ok(_)) => {
-                    let item = match this.framed.next_item(cx) {
-                        Poll::Ready(Some(Ok(el))) => el,
-                        Poll::Ready(Some(Err(err))) => {
-                            *this.state = State::FramedError(DispatcherError::Decoder(err));
-                            return true;
-                        }
-                        Poll::Pending => return false,
-                        Poll::Ready(None) => {
-                            *this.state = State::Stopping;
-                            return true;
-                        }
-                    };
-
-                    let tx = this.tx.clone();
-                    let fut = this.service.call(item);
-                    actix_rt::spawn(async move {
-                        let item = fut.await;
-                        let _ = tx.send(item.map(Message::Item));
-                    });
-                }
-                Poll::Pending => return false,
-                Poll::Ready(Err(err)) => {
-                    *this.state = State::Error(DispatcherError::Service(err));
-                    return true;
-                }
-            }
-        }
-    }
-
-    /// write to framed object
-    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
-    where
-        S: Service<<U as Decoder>::Item, Response = I>,
-        S::Error: 'static,
-        S::Future: 'static,
-        T: AsyncRead + AsyncWrite,
-        U: Decoder + Encoder<I>,
-        I: 'static,
-        <U as Encoder<I>>::Error: fmt::Debug,
-    {
-        loop {
-            let mut this = self.as_mut().project();
-            while !this.framed.is_write_buf_full() {
-                match Pin::new(&mut this.rx).poll_next(cx) {
-                    Poll::Ready(Some(Ok(Message::Item(msg)))) => {
-                        if let Err(err) = this.framed.as_mut().write(msg) {
-                            *this.state = State::FramedError(DispatcherError::Encoder(err));
-                            return true;
-                        }
-                    }
-                    Poll::Ready(Some(Ok(Message::Close))) => {
-                        *this.state = State::FlushAndStop;
-                        return true;
-                    }
-                    Poll::Ready(Some(Err(err))) => {
-                        *this.state = State::Error(DispatcherError::Service(err));
-                        return true;
-                    }
-                    Poll::Ready(None) | Poll::Pending => break,
-                }
-            }
-
-            if !this.framed.is_write_buf_empty() {
-                match this.framed.flush(cx) {
-                    Poll::Pending => break,
-                    Poll::Ready(Ok(_)) => (),
-                    Poll::Ready(Err(err)) => {
-                        debug!("Error sending data: {:?}", err);
-                        *this.state = State::FramedError(DispatcherError::Encoder(err));
-                        return true;
-                    }
-                }
-            } else {
-                break;
-            }
-        }
-
-        false
-    }
-}
-
-impl<S, T, U, I> Future for Dispatcher<S, T, U, I>
-where
-    S: Service<<U as Decoder>::Item, Response = I>,
-    S::Error: 'static,
-    S::Future: 'static,
-    T: AsyncRead + AsyncWrite,
-    U: Decoder + Encoder<I>,
-    I: 'static,
-    <U as Encoder<I>>::Error: fmt::Debug,
-    <U as Decoder>::Error: fmt::Debug,
-{
-    type Output = Result<(), DispatcherError<S::Error, U, I>>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        loop {
-            let this = self.as_mut().project();
-
-            return match this.state {
-                State::Processing => {
-                    if self.as_mut().poll_read(cx) || self.as_mut().poll_write(cx) {
-                        continue;
-                    } else {
-                        Poll::Pending
-                    }
-                }
-                State::Error(_) => {
-                    // flush write buffer
-                    if !this.framed.is_write_buf_empty() && this.framed.flush(cx).is_pending() {
-                        return Poll::Pending;
-                    }
-                    Poll::Ready(Err(this.state.take_error()))
-                }
-                State::FlushAndStop => {
-                    if !this.framed.is_write_buf_empty() {
-                        match this.framed.flush(cx) {
-                            Poll::Ready(Err(err)) => {
-                                debug!("Error sending data: {:?}", err);
-                                Poll::Ready(Ok(()))
-                            }
-                            Poll::Pending => Poll::Pending,
-                            Poll::Ready(Ok(_)) => Poll::Ready(Ok(())),
-                        }
-                    } else {
-                        Poll::Ready(Ok(()))
-                    }
-                }
-                State::FramedError(_) => Poll::Ready(Err(this.state.take_framed_error())),
-                State::Stopping => Poll::Ready(Ok(())),
-            };
-        }
-    }
-}

actix-utils/src/future/mod.rs

@@ -0,0 +1,7 @@
+//! Asynchronous values.
+
+mod poll_fn;
+mod ready;
+
+pub use self::poll_fn::{poll_fn, PollFn};
+pub use self::ready::{err, ok, ready, Ready};

actix-utils/src/future/poll_fn.rs

@@ -0,0 +1,65 @@
+//! Simple "poll function" future and factory.
+
+use core::{
+    fmt,
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+/// Create a future driven by the provided function that receives a task context.
+pub fn poll_fn<F, T>(f: F) -> PollFn<F>
+where
+    F: FnMut(&mut Context<'_>) -> Poll<T>,
+{
+    PollFn { f }
+}
+
+/// A Future driven by the inner function.
+pub struct PollFn<F> {
+    f: F,
+}
+
+impl<F> Unpin for PollFn<F> {}
+
+impl<F> fmt::Debug for PollFn<F> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("PollFn").finish()
+    }
+}
+
+impl<F, T> Future for PollFn<F>
+where
+    F: FnMut(&mut Context<'_>) -> Poll<T>,
+{
+    type Output = T;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        (self.f)(cx)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[actix_rt::test]
+    async fn test_poll_fn() {
+        let res = poll_fn(|_| Poll::Ready(42)).await;
+        assert_eq!(res, 42);
+
+        let mut i = 5;
+        let res = poll_fn(|cx| {
+            i -= 1;
+
+            if i > 0 {
+                cx.waker().wake_by_ref();
+                Poll::Pending
+            } else {
+                Poll::Ready(42)
+            }
+        })
+        .await;
+        assert_eq!(res, 42);
+    }
+}

actix-utils/src/future/ready.rs

@@ -0,0 +1,122 @@
+//! When MSRV is 1.48, replace with `core::future::Ready` and `core::future::ready()`.
+
+use core::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+/// Future for the [`ready`](ready()) function.
+///
+/// Panic will occur if polled more than once.
+///
+/// # Examples
+/// ```
+/// use actix_utils::future::ready;
+///
+/// // async
+/// # async fn run() {
+/// let a = ready(1);
+/// assert_eq!(a.await, 1);
+/// # }
+///
+/// // sync
+/// let a = ready(1);
+/// assert_eq!(a.into_inner(), 1);
+/// ```
+#[derive(Debug, Clone)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct Ready<T> {
+    val: Option<T>,
+}
+
+impl<T> Ready<T> {
+    /// Unwraps the value from this immediately ready future.
+    #[inline]
+    pub fn into_inner(mut self) -> T {
+        self.val.take().unwrap()
+    }
+}
+
+impl<T> Unpin for Ready<T> {}
+
+impl<T> Future for Ready<T> {
+    type Output = T;
+
+    #[inline]
+    fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<T> {
+        let val = self.val.take().expect("Ready polled after completion");
+        Poll::Ready(val)
+    }
+}
+
+/// Creates a future that is immediately ready with a value.
+///
+/// # Examples
+/// ```no_run
+/// use actix_utils::future::ready;
+///
+/// # async fn run() {
+/// let a = ready(1);
+/// assert_eq!(a.await, 1);
+/// # }
+///
+/// // sync
+/// let a = ready(1);
+/// assert_eq!(a.into_inner(), 1);
+/// ```
+pub fn ready<T>(val: T) -> Ready<T> {
+    Ready { val: Some(val) }
+}
+
+/// Create a future that is immediately ready with a success value.
+///
+/// # Examples
+/// ```no_run
+/// use actix_utils::future::ok;
+///
+/// # async fn run() {
+/// let a = ok::<_, ()>(1);
+/// assert_eq!(a.await, Ok(1));
+/// # }
+/// ```
+pub fn ok<T, E>(val: T) -> Ready<Result<T, E>> {
+    Ready { val: Some(Ok(val)) }
+}
+
+/// Create a future that is immediately ready with an error value.
+///
+/// # Examples
+/// ```no_run
+/// use actix_utils::future::err;
+///
+/// # async fn run() {
+/// let a = err::<(), _>(1);
+/// assert_eq!(a.await, Err(1));
+/// # }
+/// ```
+pub fn err<T, E>(err: E) -> Ready<Result<T, E>> {
+    Ready {
+        val: Some(Err(err)),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use futures_util::task::noop_waker;
+
+    use super::*;
+
+    #[test]
+    #[should_panic]
+    fn multiple_poll_panics() {
+        let waker = noop_waker();
+        let mut cx = Context::from_waker(&waker);
+
+        let mut ready = ready(1);
+        assert_eq!(Pin::new(&mut ready).poll(&mut cx), Poll::Ready(1));
+
+        // panic!
+        let _ = Pin::new(&mut ready).poll(&mut cx);
+    }
+}

actix-utils/src/lib.rs

@@ -1,12 +1,9 @@
-//! Various network related services and utilities for the Actix ecosystem.
+//! Various utilities for the Actix ecosystem.
 
 #![deny(rust_2018_idioms, nonstandard_style)]
-#![allow(clippy::type_complexity)]
+#![warn(missing_docs)]
 #![doc(html_logo_url = "https://actix.rs/img/logo.png")]
 #![doc(html_favicon_url = "https://actix.rs/favicon.ico")]
 
 pub mod counter;
-pub mod dispatcher;
+pub mod future;
-pub mod mpsc;
-pub mod task;
-pub mod timeout;

actix-utils/src/timeout.rs

@@ -1,256 +0,0 @@
-//! Service that applies a timeout to requests.
-//!
-//! If the response does not complete within the specified timeout, the response will be aborted.
-
-use core::future::Future;
-use core::marker::PhantomData;
-use core::pin::Pin;
-use core::task::{Context, Poll};
-use core::{fmt, time};
-
-use actix_rt::time::{sleep, Sleep};
-use actix_service::{IntoService, Service, Transform};
-use pin_project_lite::pin_project;
-
-/// Applies a timeout to requests.
-#[derive(Debug)]
-pub struct Timeout<E = ()> {
-    timeout: time::Duration,
-    _t: PhantomData<E>,
-}
-
-/// Timeout error
-pub enum TimeoutError<E> {
-    /// Service error
-    Service(E),
-    /// Service call timeout
-    Timeout,
-}
-
-impl<E> From<E> for TimeoutError<E> {
-    fn from(err: E) -> Self {
-        TimeoutError::Service(err)
-    }
-}
-
-impl<E: fmt::Debug> fmt::Debug for TimeoutError<E> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            TimeoutError::Service(e) => write!(f, "TimeoutError::Service({:?})", e),
-            TimeoutError::Timeout => write!(f, "TimeoutError::Timeout"),
-        }
-    }
-}
-
-impl<E: fmt::Display> fmt::Display for TimeoutError<E> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            TimeoutError::Service(e) => e.fmt(f),
-            TimeoutError::Timeout => write!(f, "Service call timeout"),
-        }
-    }
-}
-
-impl<E: PartialEq> PartialEq for TimeoutError<E> {
-    fn eq(&self, other: &TimeoutError<E>) -> bool {
-        match self {
-            TimeoutError::Service(e1) => match other {
-                TimeoutError::Service(e2) => e1 == e2,
-                TimeoutError::Timeout => false,
-            },
-            TimeoutError::Timeout => matches!(other, TimeoutError::Timeout),
-        }
-    }
-}
-
-impl<E> Timeout<E> {
-    pub fn new(timeout: time::Duration) -> Self {
-        Timeout {
-            timeout,
-            _t: PhantomData,
-        }
-    }
-}
-
-impl<E> Clone for Timeout<E> {
-    fn clone(&self) -> Self {
-        Timeout::new(self.timeout)
-    }
-}
-
-impl<S, E, Req> Transform<S, Req> for Timeout<E>
-where
-    S: Service<Req>,
-{
-    type Response = S::Response;
-    type Error = TimeoutError<S::Error>;
-    type Transform = TimeoutService<S, Req>;
-    type InitError = E;
-    type Future = TimeoutFuture<Self::Transform, Self::InitError>;
-
-    fn new_transform(&self, service: S) -> Self::Future {
-        let service = TimeoutService {
-            service,
-            timeout: self.timeout,
-            _phantom: PhantomData,
-        };
-
-        TimeoutFuture {
-            service: Some(service),
-            _err: PhantomData,
-        }
-    }
-}
-
-pub struct TimeoutFuture<T, E> {
-    service: Option<T>,
-    _err: PhantomData<E>,
-}
-
-impl<T, E> Unpin for TimeoutFuture<T, E> {}
-
-impl<T, E> Future for TimeoutFuture<T, E> {
-    type Output = Result<T, E>;
-
-    fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> {
-        Poll::Ready(Ok(self.get_mut().service.take().unwrap()))
-    }
-}
-
-/// Applies a timeout to requests.
-#[derive(Debug, Clone)]
-pub struct TimeoutService<S, Req> {
-    service: S,
-    timeout: time::Duration,
-    _phantom: PhantomData<Req>,
-}
-
-impl<S, Req> TimeoutService<S, Req>
-where
-    S: Service<Req>,
-{
-    pub fn new<U>(timeout: time::Duration, service: U) -> Self
-    where
-        U: IntoService<S, Req>,
-    {
-        TimeoutService {
-            timeout,
-            service: service.into_service(),
-            _phantom: PhantomData,
-        }
-    }
-}
-
-impl<S, Req> Service<Req> for TimeoutService<S, Req>
-where
-    S: Service<Req>,
-{
-    type Response = S::Response;
-    type Error = TimeoutError<S::Error>;
-    type Future = TimeoutServiceResponse<S, Req>;
-
-    actix_service::forward_ready!(service);
-
-    fn call(&self, request: Req) -> Self::Future {
-        TimeoutServiceResponse {
-            fut: self.service.call(request),
-            sleep: sleep(self.timeout),
-        }
-    }
-}
-
-pin_project! {
-    /// `TimeoutService` response future
-    #[derive(Debug)]
-    pub struct TimeoutServiceResponse<S, Req>
-    where
-        S: Service<Req>
-    {
-        #[pin]
-        fut: S::Future,
-        #[pin]
-        sleep: Sleep,
-    }
-}
-
-impl<S, Req> Future for TimeoutServiceResponse<S, Req>
-where
-    S: Service<Req>,
-{
-    type Output = Result<S::Response, TimeoutError<S::Error>>;
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let this = self.project();
-
-        // First, try polling the future
-        if let Poll::Ready(res) = this.fut.poll(cx) {
-            return match res {
-                Ok(v) => Poll::Ready(Ok(v)),
-                Err(e) => Poll::Ready(Err(TimeoutError::Service(e))),
-            };
-        }
-
-        // Now check the sleep
-        this.sleep.poll(cx).map(|_| Err(TimeoutError::Timeout))
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use core::task::Poll;
-    use core::time::Duration;
-
-    use super::*;
-    use actix_service::{apply, fn_factory, Service, ServiceFactory};
-    use futures_core::future::LocalBoxFuture;
-
-    struct SleepService(Duration);
-
-    impl Service<()> for SleepService {
-        type Response = ();
-        type Error = ();
-        type Future = LocalBoxFuture<'static, Result<(), ()>>;
-
-        actix_service::always_ready!();
-
-        fn call(&self, _: ()) -> Self::Future {
-            let sleep = actix_rt::time::sleep(self.0);
-            Box::pin(async move {
-                sleep.await;
-                Ok(())
-            })
-        }
-    }
-
-    #[actix_rt::test]
-    async fn test_success() {
-        let resolution = Duration::from_millis(100);
-        let wait_time = Duration::from_millis(50);
-
-        let timeout = TimeoutService::new(resolution, SleepService(wait_time));
-        assert_eq!(timeout.call(()).await, Ok(()));
-    }
-
-    #[actix_rt::test]
-    async fn test_timeout() {
-        let resolution = Duration::from_millis(100);
-        let wait_time = Duration::from_millis(500);
-
-        let timeout = TimeoutService::new(resolution, SleepService(wait_time));
-        assert_eq!(timeout.call(()).await, Err(TimeoutError::Timeout));
-    }
-
-    #[actix_rt::test]
-    async fn test_timeout_new_service() {
-        let resolution = Duration::from_millis(100);
-        let wait_time = Duration::from_millis(500);
-
-        let timeout = apply(
-            Timeout::new(resolution),
-            fn_factory(|| async { Ok::<_, ()>(SleepService(wait_time)) }),
-        );
-        let srv = timeout.new_service(&()).await.unwrap();
-
-        assert_eq!(srv.call(()).await, Err(TimeoutError::Timeout));
-    }
-}

local-channel/CHANGES.md

@@ -0,0 +1,7 @@
+# Changes
+
+## Unreleased - 2021-xx-xx
+
+
+## 0.1.1 - 2021-03-29
+* Move local mpsc channel to it's own crate.

local-channel/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "local-channel"
+version = "0.1.2"
+description = "A non-threadsafe multi-producer, single-consumer, futures-aware, FIFO queue"
+authors = [
+    "Nikolay Kim <fafhrd91@gmail.com>",
+    "Rob Ede <robjtede@icloud.com>",
+]
+repository = "https://github.com/actix/actix-net.git"
+keywords = ["channel", "local", "futures"]
+license = "MIT OR Apache-2.0"
+edition = "2018"
+
+[dependencies]
+futures-core = { version = "0.3.7", default-features = false }
+futures-sink = { version = "0.3.7", default-features = false }
+futures-util = { version = "0.3.7", default-features = false }
+local-waker = "0.1"
+
+[dev-dependencies]
+tokio = { version = "1", features = ["rt", "macros"] }

local-channel/src/lib.rs

@@ -0,0 +1,3 @@
+//! Non-thread-safe channels.
+
+pub mod mpsc;

local-channel/src/mpsc.rs

@@ -1,31 +1,35 @@
-//! A multi-producer, single-consumer, futures-aware, FIFO queue.
+//! A non-thread-safe multi-producer, single-consumer, futures-aware, FIFO queue.
 
-use core::any::Any;
+use core::{
-use core::cell::RefCell;
+    cell::RefCell,
-use core::fmt;
+    fmt,
-use core::pin::Pin;
+    pin::Pin,
-use core::task::{Context, Poll};
+    task::{Context, Poll},
+};
 
-use std::collections::VecDeque;
+use std::{collections::VecDeque, error::Error, rc::Rc};
-use std::error::Error;
-use std::rc::Rc;
 
 use futures_core::stream::Stream;
 use futures_sink::Sink;
-
+use futures_util::future::poll_fn;
-use crate::task::LocalWaker;
+use local_waker::LocalWaker;
 
 /// Creates a unbounded in-memory channel with buffered storage.
+///
+/// [Sender]s and [Receiver]s are `!Send`.
 pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
     let shared = Rc::new(RefCell::new(Shared {
         has_receiver: true,
         buffer: VecDeque::new(),
         blocked_recv: LocalWaker::new(),
     }));
+
     let sender = Sender {
         shared: shared.clone(),
     };
+
     let receiver = Receiver { shared };
+
     (sender, receiver)
 }
 
@@ -50,18 +54,22 @@ impl<T> Sender<T> {
     /// Sends the provided message along this channel.
     pub fn send(&self, item: T) -> Result<(), SendError<T>> {
         let mut shared = self.shared.borrow_mut();
+
         if !shared.has_receiver {
-            return Err(SendError(item)); // receiver was dropped
+            // receiver was dropped
+            return Err(SendError(item));
         };
+
         shared.buffer.push_back(item);
         shared.blocked_recv.wake();
+
         Ok(())
     }
 
-    /// Closes the sender half
+    /// Closes the sender half.
     ///
-    /// This prevents any further messages from being sent on the channel while
+    /// This prevents any further messages from being sent on the channel, by any sender, while
-    /// still enabling the receiver to drain messages that are buffered.
+    /// still enabling the receiver to drain messages that are already buffered.
     pub fn close(&mut self) {
         self.shared.borrow_mut().has_receiver = false;
     }
@@ -110,14 +118,24 @@ impl<T> Drop for Sender<T> {
 
 /// The receiving end of a channel which implements the `Stream` trait.
 ///
-/// This is created by the `channel` function.
+/// This is created by the [`channel`] function.
 #[derive(Debug)]
 pub struct Receiver<T> {
     shared: Rc<RefCell<Shared<T>>>,
 }
 
 impl<T> Receiver<T> {
-    /// Create Sender
+    /// Receive the next value.
+    ///
+    /// Returns `None` if the channel is empty and has been [closed](Sender::close) explicitly or
+    /// when all senders have been dropped and, therefore, no more values can ever be sent though
+    /// this channel.
+    pub async fn recv(&mut self) -> Option<T> {
+        let mut this = Pin::new(self);
+        poll_fn(|cx| this.as_mut().poll_next(cx)).await
+    }
+
+    /// Create an associated [Sender].
     pub fn sender(&self) -> Sender<T> {
         Sender {
             shared: self.shared.clone(),
@@ -132,11 +150,13 @@ impl<T> Stream for Receiver<T> {
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut shared = self.shared.borrow_mut();
+
         if Rc::strong_count(&self.shared) == 1 {
-            // All senders have been dropped, so drain the buffer and end the
+            // All senders have been dropped, so drain the buffer and end the stream.
-            // stream.
+            return Poll::Ready(shared.buffer.pop_front());
-            Poll::Ready(shared.buffer.pop_front())
+        }
-        } else if let Some(msg) = shared.buffer.pop_front() {
+
+        if let Some(msg) = shared.buffer.pop_front() {
             Poll::Ready(Some(msg))
         } else {
             shared.blocked_recv.register(cx.waker());
@@ -153,9 +173,17 @@ impl<T> Drop for Receiver<T> {
     }
 }
 
-/// Error type for sending, used when the receiving end of a channel is
+/// Error returned when attempting to send after the channels' [Receiver] is dropped or closed.
-/// dropped
+///
-pub struct SendError<T>(T);
+/// Allows access to message that failed to send with [`into_inner`](Self::into_inner).
+pub struct SendError<T>(pub T);
+
+impl<T> SendError<T> {
+    /// Returns the message that was attempted to be sent but failed.
+    pub fn into_inner(self) -> T {
+        self.0
+    }
+}
 
 impl<T> fmt::Debug for SendError<T> {
     fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
@@ -169,26 +197,15 @@ impl<T> fmt::Display for SendError<T> {
     }
 }
 
-impl<T: Any> Error for SendError<T> {
+impl<T> Error for SendError<T> {}
-    fn description(&self) -> &str {
-        "send failed because receiver is gone"
-    }
-}
-
-impl<T> SendError<T> {
-    /// Returns the message that was attempted to be sent but failed.
-    pub fn into_inner(self) -> T {
-        self.0
-    }
-}
 
 #[cfg(test)]
 mod tests {
-    use super::*;
+    use futures_util::{future::lazy, StreamExt as _};
-    use futures_util::future::lazy;
-    use futures_util::{stream::Stream, StreamExt};
 
-    #[actix_rt::test]
+    use super::*;
+
+    #[tokio::test]
     async fn test_mpsc() {
         let (tx, mut rx) = channel();
         tx.send("test").unwrap();
@@ -221,4 +238,18 @@ mod tests {
         assert!(tx.send("test").is_err());
         assert!(tx2.send("test").is_err());
     }
+
+    #[tokio::test]
+    async fn test_recv() {
+        let (tx, mut rx) = channel();
+        tx.send("test").unwrap();
+        assert_eq!(rx.recv().await.unwrap(), "test");
+        drop(tx);
+
+        let (tx, mut rx) = channel();
+        tx.send("test").unwrap();
+        assert_eq!(rx.recv().await.unwrap(), "test");
+        drop(tx);
+        assert!(rx.recv().await.is_none());
+    }
 }

local-waker/CHANGES.md

@@ -0,0 +1,11 @@
+# Changes
+
+## Unreleased - 2021-xx-xx
+
+
+## 0.1.2 - 2021-04-01
+* Fix crate metadata.
+
+
+## 0.1.1 - 2021-03-29
+* Move `LocalWaker` to it's own crate.

local-waker/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "local-waker"
+version = "0.1.1"
+description = "A synchronization primitive for thread-local task wakeup"
+authors = [
+    "Nikolay Kim <fafhrd91@gmail.com>",
+    "Rob Ede <robjtede@icloud.com>",
+]
+keywords = ["waker", "local", "futures", "no-std"]
+repository = "https://github.com/actix/actix-net.git"
+categories = ["asynchronous", "no-std"]
+license = "MIT OR Apache-2.0"
+edition = "2018"
+
+[dependencies]

local-waker/src/lib.rs

@@ -1,3 +1,9 @@
+//! A synchronization primitive for thread-local task wakeup.
+//!
+//! See docs for [`LocalWaker`].
+
+#![no_std]
+
 use core::{cell::Cell, fmt, marker::PhantomData, task::Waker};
 
 /// A synchronization primitive for task wakeup.
@@ -9,11 +15,14 @@ use core::{cell::Cell, fmt, marker::PhantomData, task::Waker};
 /// logical task.
 ///
 /// Consumers should call [`register`] before checking the result of a computation and producers
-/// should call `wake` after producing the computation (this differs from the usual `thread::park`
+/// should call [`wake`] after producing the computation (this differs from the usual `thread::park`
 /// pattern). It is also permitted for [`wake`] to be called _before_ [`register`]. This results in
 /// a no-op.
 ///
 /// A single `LocalWaker` may be reused for any number of calls to [`register`] or [`wake`].
+///
+/// [`register`]: LocalWaker::register
+/// [`wake`]: LocalWaker::wake
 #[derive(Default)]
 pub struct LocalWaker {
     pub(crate) waker: Cell<Option<Waker>>,