Merge pull request #109 from JohnTitor/new-tls

actix-tls: Bump up to 2.0.0-alpha.1

.appveyor.yml

@@ -1,41 +0,0 @@
-environment:
-  global:
-    PROJECT_NAME: actix-net
-  matrix:
-    # Stable channel
-    - TARGET: i686-pc-windows-msvc
-      CHANNEL: stable
-    - TARGET: x86_64-pc-windows-gnu
-      CHANNEL: stable
-    - TARGET: x86_64-pc-windows-msvc
-      CHANNEL: stable
-    # Nightly channel
-    - TARGET: i686-pc-windows-msvc
-      CHANNEL: nightly
-    - TARGET: x86_64-pc-windows-gnu
-      CHANNEL: nightly
-    - TARGET: x86_64-pc-windows-msvc
-      CHANNEL: nightly
-
-# Install Rust and Cargo
-# (Based on from https://github.com/rust-lang/libc/blob/master/appveyor.yml)
-install:
-  - ps: >-
-        If ($Env:TARGET -eq 'x86_64-pc-windows-gnu') {
-           $Env:PATH += ';C:\msys64\mingw64\bin'
-        } ElseIf ($Env:TARGET -eq 'i686-pc-windows-gnu') {
-           $Env:PATH += ';C:\MinGW\bin'
-        }
-  - curl -sSf -o rustup-init.exe https://win.rustup.rs
-  - rustup-init.exe --default-host %TARGET% --default-toolchain %CHANNEL% -y
-  - set PATH=%PATH%;C:\Users\appveyor\.cargo\bin
-  - rustc -Vv
-  - cargo -V
-
-# 'cargo test' takes care of building for us, so disable Appveyor's build stage.
-build: false
-
-# Equivalent to Travis' `script` phase
-test_script:
-  - cargo clean
-  - cargo test

.github/workflows/bench.yml

@@ -0,0 +1,23 @@
+name: Benchmark (Linux)
+
+on: [push, pull_request]
+
+jobs:
+  check_benchmark:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@master
+
+      - name: Install Rust
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: nightly
+          profile: minimal
+          override: true
+
+      - name: Check benchmark
+        uses: actions-rs/cargo@v1
+        with:
+          command: bench
+          args: --package=actix-service

.github/workflows/clippy.yml

@@ -0,0 +1,18 @@
+on: pull_request
+
+name: Clippy Check
+jobs:
+  clippy_check:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@master
+      - uses: actions-rs/toolchain@v1
+        with:
+          toolchain: nightly
+          components: clippy
+          profile: minimal
+          override: true
+      - uses: actions-rs/clippy-check@v1
+        with:
+          token: ${{ secrets.GITHUB_TOKEN }}
+          args: --all-features --all --tests

.github/workflows/linux.yml

@@ -0,0 +1,77 @@
+name: CI (Linux)
+
+on: [push, pull_request]
+
+jobs:
+  build_and_test:
+    strategy:
+      fail-fast: false
+      matrix:
+        version:
+          - 1.39.0
+          - stable
+          - nightly
+
+    name: ${{ matrix.version }} - x86_64-unknown-linux-gnu
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@master
+
+      - name: Install ${{ matrix.version }}
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: ${{ matrix.version }}-x86_64-unknown-linux-gnu
+          profile: minimal
+          override: true
+
+      - name: Generate Cargo.lock
+        uses: actions-rs/cargo@v1
+        with:
+          command: generate-lockfile
+      - name: Cache cargo registry
+        uses: actions/cache@v1
+        with:
+          path: ~/.cargo/registry
+          key: ${{ matrix.version }}-x86_64-unknown-linux-gnu-cargo-registry-trimmed-${{ hashFiles('**/Cargo.lock') }}
+      - name: Cache cargo index
+        uses: actions/cache@v1
+        with:
+          path: ~/.cargo/git
+          key: ${{ matrix.version }}-x86_64-unknown-linux-gnu-cargo-index-trimmed-${{ hashFiles('**/Cargo.lock') }}
+      - name: Cache cargo build
+        uses: actions/cache@v1
+        with:
+          path: target
+          key: ${{ matrix.version }}-x86_64-unknown-linux-gnu-cargo-build-trimmed-${{ hashFiles('**/Cargo.lock') }}
+
+      - name: check build
+        uses: actions-rs/cargo@v1
+        with:
+          command: check
+          args: --all --bins --examples --tests
+
+      - name: tests
+        uses: actions-rs/cargo@v1
+        timeout-minutes: 40
+        with:
+          command: test
+          args: --all --all-features --no-fail-fast -- --nocapture
+
+      - name: Generate coverage file
+        if: matrix.version == 'stable' && github.ref == 'refs/heads/master'
+        run: |
+          cargo install cargo-tarpaulin
+          cargo tarpaulin --out Xml --workspace --all-features
+
+      - name: Upload to Codecov
+        if: matrix.version == 'stable' && github.ref == 'refs/heads/master'
+        uses: codecov/codecov-action@v1
+        with:
+          token: ${{ secrets.CODECOV_TOKEN }}
+          file: cobertura.xml
+
+      - name: Clear the cargo caches
+        run: |
+          cargo install cargo-cache --no-default-features --features ci-autoclean
+          cargo-cache

.github/workflows/macos.yml

@@ -0,0 +1,37 @@
+name: CI (macOS)
+
+on: [push, pull_request]
+
+jobs:
+  build_and_test:
+    strategy:
+      fail-fast: false
+      matrix:
+        version:
+          - stable
+          - nightly
+
+    name: ${{ matrix.version }} - x86_64-apple-darwin
+    runs-on: macos-latest
+
+    steps:
+      - uses: actions/checkout@master
+
+      - name: Install ${{ matrix.version }}
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: ${{ matrix.version }}-x86_64-apple-darwin
+          profile: minimal
+          override: true
+
+      - name: check build
+        uses: actions-rs/cargo@v1
+        with:
+          command: check
+          args: --all --bins --examples --tests
+
+      - name: tests
+        uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --all --all-features --no-fail-fast -- --nocapture

.github/workflows/main.yml

@@ -1,78 +0,0 @@
-name: CI
-
-on: [push, pull_request]
-
-env:
-  VCPKGRS_DYNAMIC: 1
-
-jobs:
-  build_and_test:
-    strategy:
-      fail-fast: false
-      matrix:
-        toolchain:
-          - x86_64-pc-windows-msvc
-          - x86_64-pc-windows-gnu
-          - i686-pc-windows-msvc
-          - x86_64-apple-darwin
-        version:
-          - stable
-          - nightly
-        include:
-        - toolchain: x86_64-pc-windows-msvc
-          os: windows-latest
-          arch: x64
-        - toolchain: x86_64-pc-windows-gnu
-          os: windows-latest
-        - toolchain: i686-pc-windows-msvc
-          os: windows-latest
-          arch: x86
-        - toolchain: x86_64-apple-darwin
-          os: macOS-latest
-
-    name: ${{ matrix.version }} - ${{ matrix.toolchain }}
-    runs-on: ${{ matrix.os }}
-
-    steps:
-      - uses: actions/checkout@master
-
-      - name: Install ${{ matrix.version }}
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: ${{ matrix.version }}-${{ matrix.toolchain }}
-          default: true
-
-
-      - name: Install OpenSSL
-        if: matrix.toolchain == 'x86_64-pc-windows-msvc' || matrix.toolchain == 'i686-pc-windows-msvc'
-        run: |
-          vcpkg integrate install
-          vcpkg install openssl:${{ matrix.arch }}-windows
-
-      - name: check nightly
-        if: matrix.version == 'nightly'
-        uses: actions-rs/cargo@v1
-        with:
-          command: check
-          args: --all --benches --bins --examples --tests
-
-      - name: check stable
-        if: matrix.version == 'stable'
-        uses: actions-rs/cargo@v1
-        with:
-          command: check
-          args: --all --bins --examples --tests
-
-      - name: tests
-        if: matrix.toolchain != 'x86_64-pc-windows-gnu'
-        uses: actions-rs/cargo@v1
-        with:
-          command: test
-          args: --all --all-features -- --nocapture
-
-      - name: tests on x86_64-pc-windows-gnu
-        if: matrix.toolchain == 'x86_64-pc-windows-gnu'
-        uses: actions-rs/cargo@v1
-        with:
-          command: test
-          args: --all -- --nocapture

.github/workflows/windows-mingw.yml

@@ -0,0 +1,54 @@
+name: CI (Windows-mingw)
+
+on: [push, pull_request]
+
+env:
+  OPENSSL_DIR: d:\a\_temp\msys\msys64\usr
+
+jobs:
+  build_and_test:
+    strategy:
+      fail-fast: false
+      matrix:
+        version:
+          - stable
+          - nightly
+
+    name: ${{ matrix.version }} - x86_64-pc-windows-gnu
+    runs-on: windows-latest
+
+    steps:
+      - uses: actions/checkout@master
+
+      - name: Install ${{ matrix.version }}
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: ${{ matrix.version }}-x86_64-pc-windows-gnu
+          profile: minimal
+          override: true
+
+      - name: Install MSYS2
+        uses: numworks/setup-msys2@v1
+
+      - name: Install OpenSSL
+        run: |
+          msys2do pacman --noconfirm -S openssl-devel pkg-config
+
+      - name: Copy and check libs
+        run: |
+          Copy-Item d:\a\_temp\msys\msys64\usr\lib\libssl.dll.a d:\a\_temp\msys\msys64\usr\lib\libssl.dll
+          Copy-Item d:\a\_temp\msys\msys64\usr\lib\libcrypto.dll.a d:\a\_temp\msys\msys64\usr\lib\libcrypto.dll
+          Get-ChildItem d:\a\_temp\msys\msys64\usr\lib
+          Get-ChildItem d:\a\_temp\msys\msys64\usr
+
+      - name: check build
+        uses: actions-rs/cargo@v1
+        with:
+          command: check
+          args: --all --bins --examples --tests
+
+      - name: tests
+        uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --all --all-features --no-fail-fast -- --nocapture

.github/workflows/windows.yml

@@ -0,0 +1,63 @@
+name: CI (Windows)
+
+on: [push, pull_request]
+
+env:
+  VCPKGRS_DYNAMIC: 1
+
+jobs:
+  build_and_test:
+    strategy:
+      fail-fast: false
+      matrix:
+        version:
+          - stable
+          - nightly
+        target:
+          - x86_64-pc-windows-msvc
+          - i686-pc-windows-msvc
+
+    name: ${{ matrix.version }} - ${{ matrix.target }}
+    runs-on: windows-latest
+
+    steps:
+      - uses: actions/checkout@master
+
+      - name: Install ${{ matrix.version }}
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: ${{ matrix.version }}-${{ matrix.target }}
+          profile: minimal
+          override: true
+
+      - name: Install OpenSSL (x64)
+        if: matrix.target == 'x86_64-pc-windows-msvc'
+        run: |
+          vcpkg integrate install
+          vcpkg install openssl:x64-windows
+          Get-ChildItem C:\vcpkg\installed\x64-windows\bin
+          Get-ChildItem C:\vcpkg\installed\x64-windows\lib
+          Copy-Item C:\vcpkg\installed\x64-windows\bin\libcrypto-1_1-x64.dll C:\vcpkg\installed\x64-windows\bin\libcrypto.dll
+          Copy-Item C:\vcpkg\installed\x64-windows\bin\libssl-1_1-x64.dll C:\vcpkg\installed\x64-windows\bin\libssl.dll
+
+      - name: Install OpenSSL (x86)
+        if: matrix.target == 'i686-pc-windows-msvc'
+        run: |
+          vcpkg integrate install
+          vcpkg install openssl:x86-windows
+          Get-ChildItem C:\vcpkg\installed\x86-windows\bin
+          Get-ChildItem C:\vcpkg\installed\x86-windows\lib
+          Copy-Item C:\vcpkg\installed\x86-windows\bin\libcrypto-1_1.dll C:\vcpkg\installed\x86-windows\bin\libcrypto.dll
+          Copy-Item C:\vcpkg\installed\x86-windows\bin\libssl-1_1.dll C:\vcpkg\installed\x86-windows\bin\libssl.dll
+
+      - name: check build
+        uses: actions-rs/cargo@v1
+        with:
+          command: check
+          args: --all --bins --examples --tests
+
+      - name: tests
+        uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --all --all-features --no-fail-fast -- --nocapture

.travis.yml

@@ -1,49 +0,0 @@
-language: rust
-sudo: required
-dist: trusty
-
-cache:
-  cargo: true
-  apt: true
-
-matrix:
-  include:
-    - rust: stable
-    - rust: beta
-    - rust: nightly-2019-11-07
-  allow_failures:
-    - rust: nightly-2019-11-07
-
-env:
-  global:
-    - RUSTFLAGS="-C link-dead-code"
-    - OPENSSL_VERSION=openssl-1.0.2
-
-before_install:
-  - sudo add-apt-repository -y ppa:0k53d-karl-f830m/openssl
-  - sudo apt-get update -qq
-  - sudo apt-get install -y openssl libssl-dev libelf-dev libdw-dev cmake gcc binutils-dev libiberty-dev
-
-before_cache: |
-  if [[ "$TRAVIS_RUST_VERSION" == "nightly-2019-11-07" ]]; then
-    RUSTFLAGS="--cfg procmacro2_semver_exempt" cargo install --version 0.6.11 cargo-tarpaulin
-  fi
-
-# Add clippy
-before_script:
-  - export PATH=$PATH:~/.cargo/bin
-
-script:
-  - |
-    if [[ "$TRAVIS_RUST_VERSION" != "nightly-2019-11-07" ]]; then
-    cargo clean
-    cargo test --all --all-features -- --nocapture
-    fi
-
-after_success:
-  - |
-    if [[ "$TRAVIS_RUST_VERSION" == "nightly-2019-11-07" ]]; then
-    taskset -c 0 cargo tarpaulin --all --all-features --out Xml
-    echo "Uploaded code coverage"
-    bash <(curl -s https://codecov.io/bash)
-    fi

Cargo.toml

@@ -10,6 +10,7 @@ members = [
   "actix-testing",
   "actix-threadpool",
   "actix-tls",
+  "actix-tracing",
   "actix-utils",
   "router",
   "string",
@@ -26,6 +27,7 @@ actix-service = { path = "actix-service" }
 actix-testing = { path = "actix-testing" }
 actix-threadpool = { path = "actix-threadpool" }
 actix-tls = { path = "actix-tls" }
+actix-tracing = { path = "actix-tracing" }
 actix-utils = { path = "actix-utils" }
 actix-router = { path = "router" }
 bytestring = { path = "string" }

README.md

@@ -1,7 +1,16 @@
-# Actix net [![Build Status](https://travis-ci.org/actix/actix-net.svg?branch=master)](https://travis-ci.org/actix/actix-net) [![codecov](https://codecov.io/gh/actix/actix-net/branch/master/graph/badge.svg)](https://codecov.io/gh/actix/actix-net) [![Join the chat at https://gitter.im/actix/actix](https://badges.gitter.im/actix/actix.svg)](https://gitter.im/actix/actix?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
+# Actix net [![codecov](https://codecov.io/gh/actix/actix-net/branch/master/graph/badge.svg)](https://codecov.io/gh/actix/actix-net) [![Join the chat at https://gitter.im/actix/actix](https://badges.gitter.im/actix/actix.svg)](https://gitter.im/actix/actix?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
 
 Actix net - framework for composable network services
 
+## Build statuses
+
+| Platform         | Build Status |
+| ---------------- | ------------ |
+| Linux            | [![build status](https://github.com/actix/actix-net/workflows/CI%20%28Linux%29/badge.svg?branch=master&event=push)](https://github.com/actix/actix-net/actions?query=workflow%3A"CI+(Linux)") |
+| macOS            | [![build status](https://github.com/actix/actix-net/workflows/CI%20%28macOS%29/badge.svg?branch=master&event=push)](https://github.com/actix/actix-net/actions?query=workflow%3A"CI+(macOS)") |
+| Windows          | [![build status](https://github.com/actix/actix-net/workflows/CI%20%28Windows%29/badge.svg?branch=master&event=push)](https://github.com/actix/actix-net/actions?query=workflow%3A"CI+(Windows)") |
+| Windows (MinGW)  | [![build status](https://github.com/actix/actix-net/workflows/CI%20%28Windows-mingw%29/badge.svg?branch=master&event=push)](https://github.com/actix/actix-net/actions?query=workflow%3A"CI+(Windows-mingw)") |
+
 ## Documentation & community resources
 
 * [Chat on gitter](https://gitter.im/actix/actix)

actix-connect/CHANGES.md

@@ -1,5 +1,16 @@
 # Changes
 
+## [2.0.0-alpha.1] - 2020-03-03
+
+### Changed
+
+* Update `rustls` dependency to 0.17
+* Update `tokio-rustls` dependency to 0.13
+
+## [1.0.2] - 2020-01-15
+
+* Fix actix-service 1.0.3 compatibility
+
 ## [1.0.1] - 2019-12-15
 
 * Fix trust-dns-resolver compilation

actix-connect/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-connect"
-version = "1.0.1"
+version = "2.0.0-alpha.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix connect - tcp connector service"
 keywords = ["network", "framework", "async", "futures"]
@@ -10,7 +10,6 @@ documentation = "https://docs.rs/actix-connect/"
 categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
 edition = "2018"
-workspace = ".."
 
 [package.metadata.docs.rs]
 features = ["openssl", "rustls", "uri"]
@@ -32,12 +31,12 @@ rustls = ["rust-tls", "tokio-rustls", "webpki"]
 uri = ["http"]
 
 [dependencies]
-actix-service = "1.0.0"
+actix-service = "1.0.3"
 actix-codec = "0.2.0"
-actix-utils = "1.0.3"
+actix-utils = "1.0.6"
 actix-rt = "1.0.0"
 derive_more = "0.99.2"
-either = "1.5.2"
+either = "1.5.3"
 futures = "0.3.1"
 http = { version = "0.2.0", optional = true }
 log = "0.4"
@@ -49,8 +48,8 @@ open-ssl = { version="0.10", package = "openssl", optional = true }
 tokio-openssl = { version = "0.4.0", optional = true }
 
 # rustls
-rust-tls = { version = "0.16.0", package = "rustls", optional = true }
-tokio-rustls = { version = "0.12.0", optional = true }
+rust-tls = { version = "0.17.0", package = "rustls", optional = true }
+tokio-rustls = { version = "0.13.0", optional = true }
 webpki = { version = "0.21", optional = true }
 
 [dev-dependencies]

actix-connect/src/lib.rs

@@ -72,7 +72,7 @@ pub fn start_default_resolver() -> AsyncResolver {
 }
 
 /// Create tcp connector service
-pub fn new_connector<T: Address>(
+pub fn new_connector<T: Address + 'static>(
     resolver: AsyncResolver,
 ) -> impl Service<Request = Connect<T>, Response = Connection<T, TcpStream>, Error = ConnectError>
        + Clone {
@@ -80,7 +80,7 @@ pub fn new_connector<T: Address>(
 }
 
 /// Create tcp connector service
-pub fn new_connector_factory<T: Address>(
+pub fn new_connector_factory<T: Address + 'static>(
     resolver: AsyncResolver,
 ) -> impl ServiceFactory<
     Config = (),
@@ -93,14 +93,14 @@ pub fn new_connector_factory<T: Address>(
 }
 
 /// Create connector service with default parameters
-pub fn default_connector<T: Address>(
+pub fn default_connector<T: Address + 'static>(
 ) -> impl Service<Request = Connect<T>, Response = Connection<T, TcpStream>, Error = ConnectError>
        + Clone {
     pipeline(Resolver::default()).and_then(TcpConnector::new())
 }
 
 /// Create connector service factory with default parameters
-pub fn default_connector_factory<T: Address>() -> impl ServiceFactory<
+pub fn default_connector_factory<T: Address + 'static>() -> impl ServiceFactory<
     Config = (),
     Request = Connect<T>,
     Response = Connection<T, TcpStream>,

actix-connect/src/ssl/rustls.rs

@@ -38,7 +38,7 @@ where
 {
     pub fn service(connector: Arc<ClientConfig>) -> RustlsConnectorService<T, U> {
         RustlsConnectorService {
-            connector: connector,
+            connector,
             _t: PhantomData,
         }
     }

actix-ioframe/Cargo.toml

@@ -27,5 +27,5 @@ pin-project = "0.4.6"
 log = "0.4"
 
 [dev-dependencies]
-actix-connect = "1.0.0"
+actix-connect = "2.0.0-alpha.1"
 actix-testing = "1.0.0"

actix-ioframe/src/dispatcher.rs

@@ -213,7 +213,7 @@ where
                 // drain service responses
                 match Pin::new(&mut self.rx).poll_next(cx) {
                     Poll::Ready(Some(Ok(msg))) => {
-                        if let Err(_) = self.framed.write(msg) {
+                        if self.framed.write(msg).is_err() {
                             return Poll::Ready(Ok(()));
                         }
                     }

actix-macros/src/lib.rs

@@ -14,6 +14,7 @@ use quote::quote;
 ///     println!("Hello world");
 /// }
 /// ```
+#[allow(clippy::needless_doctest_main)]
 #[proc_macro_attribute]
 #[cfg(not(test))] // Work around for rust-lang/rust#62127
 pub fn main(_: TokenStream, item: TokenStream) -> TokenStream {

actix-rt/CHANGES.md

@@ -1,5 +1,11 @@
 # Changes
 
+## [TBD] - [TBD]
+
+- Expose `System::is_set` to check if current system is running
+
+- Add `Arbiter::local_join` associated function to get be able to `await` for spawned futures
+
 ## [1.0.0] - 2019-12-11
 
 * Update dependencies

actix-rt/Cargo.toml

@@ -18,6 +18,7 @@ path = "src/lib.rs"
 [dependencies]
 actix-macros = "0.1.0"
 actix-threadpool = "0.3"
-futures = "0.3.1"
+futures-channel = { version = "0.3.1", default-features = false }
+futures-util = { version = "0.3.1", default-features = false }
 copyless = "0.1.4"
-tokio = { version = "0.2.6", default-features=false, features = ["rt-core", "rt-util", "io-driver", "tcp", "uds", "udp", "time", "signal", "stream"] }
+tokio = { version = "0.2.6", default-features = false, features = ["rt-core", "rt-util", "io-driver", "tcp", "uds", "udp", "time", "signal", "stream"] }

actix-rt/src/arbiter.rs

@@ -6,19 +6,22 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::task::{Context, Poll};
 use std::{fmt, thread};
 
-use futures::channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
-use futures::channel::oneshot::{channel, Canceled, Sender};
-use futures::{future, Future, FutureExt, Stream};
+use futures_channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
+use futures_channel::oneshot::{channel, Canceled, Sender};
+use futures_util::{future::{self, Future, FutureExt}, stream::Stream};
 
 use crate::runtime::Runtime;
 use crate::system::System;
 
 use copyless::BoxHelper;
 
+pub use tokio::task::JoinHandle;
+
 thread_local!(
     static ADDR: RefCell<Option<Arbiter>> = RefCell::new(None);
     static RUNNING: Cell<bool> = Cell::new(false);
     static Q: RefCell<Vec<Pin<Box<dyn Future<Output = ()>>>>> = RefCell::new(Vec::new());
+    static PENDING: RefCell<Vec<JoinHandle<()>>> = RefCell::new(Vec::new());
     static STORAGE: RefCell<HashMap<TypeId, Box<dyn Any>>> = RefCell::new(HashMap::new());
 );
 
@@ -42,8 +45,8 @@ impl fmt::Debug for ArbiterCommand {
 
 #[derive(Debug)]
 /// Arbiters provide an asynchronous execution environment for actors, functions
-/// and futures. When an Arbiter is created, they spawn a new OS thread, and
-/// host an event loop. Some Arbiter functions execute on the current thread.
+/// and futures. When an Arbiter is created, it spawns a new OS thread, and
+/// hosts an event loop. Some Arbiter functions execute on the current thread.
 pub struct Arbiter {
     sender: UnboundedSender<ArbiterCommand>,
     thread_handle: Option<thread::JoinHandle<()>>,
@@ -170,7 +173,9 @@ impl Arbiter {
         RUNNING.with(move |cell| {
             if cell.get() {
                 // Spawn the future on running executor
-                tokio::task::spawn_local(future);
+                PENDING.with(move |cell| {
+                    cell.borrow_mut().push(tokio::task::spawn_local(future));
+                })
             } else {
                 // Box the future and push it to the queue, this results in double boxing
                 // because the executor boxes the future again, but works for now
@@ -294,6 +299,15 @@ impl Arbiter {
             Ok(())
         }
     }
+
+    /// Returns a future that will be completed once all currently spawned futures
+    /// have completed.
+    pub fn local_join() -> impl Future<Output = ()> {
+        PENDING.with(move |cell| {
+            let current = cell.replace(Vec::new());
+            future::join_all(current).map(|_| ())
+        })
+    }
 }
 
 struct ArbiterController {
@@ -329,7 +343,9 @@ impl Future for ArbiterController {
                         return Poll::Ready(());
                     }
                     ArbiterCommand::Execute(fut) => {
-                        tokio::task::spawn_local(fut);
+                        PENDING.with(move |cell| {
+                            cell.borrow_mut().push(tokio::task::spawn_local(fut));
+                        });
                     }
                     ArbiterCommand::ExecuteFn(f) => {
                         f.call_box();

actix-rt/src/builder.rs

@@ -1,9 +1,9 @@
 use std::borrow::Cow;
 use std::io;
 
-use futures::channel::mpsc::unbounded;
-use futures::channel::oneshot::{channel, Receiver};
-use futures::future::{lazy, Future, FutureExt};
+use futures_channel::mpsc::unbounded;
+use futures_channel::oneshot::{channel, Receiver};
+use futures_util::future::{lazy, Future, FutureExt};
 use tokio::task::LocalSet;
 
 use crate::arbiter::{Arbiter, SystemArbiter};

actix-rt/src/lib.rs

@@ -25,7 +25,7 @@ pub use actix_threadpool as blocking;
 /// This function panics if actix system is not running.
 pub fn spawn<F>(f: F)
 where
-    F: futures::Future<Output = ()> + 'static,
+    F: futures_util::future::Future<Output = ()> + 'static,
 {
     if !System::is_set() {
         panic!("System is not running");

actix-rt/src/runtime.rs

@@ -86,7 +86,6 @@ impl Runtime {
     where
         F: Future + 'static,
     {
-        let res = self.local.block_on(&mut self.rt, f);
-        res
+        self.local.block_on(&mut self.rt, f)
     }
 }

actix-rt/src/system.rs

@@ -3,7 +3,7 @@ use std::future::Future;
 use std::io;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
-use futures::channel::mpsc::UnboundedSender;
+use futures_channel::mpsc::UnboundedSender;
 use tokio::task::LocalSet;
 
 use crate::arbiter::{Arbiter, SystemCommand};
@@ -79,8 +79,8 @@ impl System {
         })
     }
 
-    /// Set current running system.
-    pub(crate) fn is_set() -> bool {
+    /// Check if current system is running.
+    pub fn is_set() -> bool {
         CURRENT.with(|cell| cell.borrow().is_some())
     }
 

actix-rt/tests/wait_spawned.rs

@@ -0,0 +1,100 @@
+use std::time::{Duration, Instant};
+
+#[test]
+fn await_for_timer() {
+    let time = Duration::from_secs(2);
+    let instant = Instant::now();
+    actix_rt::System::new("test_wait_timer").block_on(async move {
+        tokio::time::delay_for(time).await;
+    });
+    assert!(
+        instant.elapsed() >= time,
+        "Block on should poll awaited future to completion"
+    );
+}
+
+#[test]
+fn join_another_arbiter() {
+    let time = Duration::from_secs(2);
+    let instant = Instant::now();
+    actix_rt::System::new("test_join_another_arbiter").block_on(async move {
+        let mut arbiter = actix_rt::Arbiter::new();
+        arbiter.send(Box::pin(async move {
+            tokio::time::delay_for(time).await;
+            actix_rt::Arbiter::current().stop();
+        }));
+        arbiter.join().unwrap();
+    });
+    assert!(
+        instant.elapsed() >= time,
+        "Join on another arbiter should complete only when it calls stop"
+    );
+
+    let instant = Instant::now();
+    actix_rt::System::new("test_join_another_arbiter").block_on(async move {
+        let mut arbiter = actix_rt::Arbiter::new();
+        arbiter.exec_fn(move || {
+            actix_rt::spawn(async move {
+                tokio::time::delay_for(time).await;
+                actix_rt::Arbiter::current().stop();
+            });
+        });
+        arbiter.join().unwrap();
+    });
+    assert!(
+        instant.elapsed() >= time,
+        "Join on a arbiter that has used actix_rt::spawn should wait for said future"
+    );
+
+    let instant = Instant::now();
+    actix_rt::System::new("test_join_another_arbiter").block_on(async move {
+        let mut arbiter = actix_rt::Arbiter::new();
+        arbiter.send(Box::pin(async move {
+            tokio::time::delay_for(time).await;
+            actix_rt::Arbiter::current().stop();
+        }));
+        arbiter.stop();
+        arbiter.join().unwrap();
+    });
+    assert!(
+        instant.elapsed() < time,
+        "Premature stop of arbiter should conclude regardless of it's current state"
+    );
+}
+
+#[test]
+fn join_current_arbiter() {
+    let time = Duration::from_secs(2);
+
+    let instant = Instant::now();
+    actix_rt::System::new("test_join_current_arbiter").block_on(async move {
+        actix_rt::spawn(async move {
+            tokio::time::delay_for(time).await;
+            actix_rt::Arbiter::current().stop();
+        });
+        actix_rt::Arbiter::local_join().await;
+    });
+    assert!(
+        instant.elapsed() >= time,
+        "Join on current arbiter should wait for all spawned futures"
+    );
+
+    let large_timer = Duration::from_secs(20);
+    let instant = Instant::now();
+    actix_rt::System::new("test_join_current_arbiter").block_on(async move {
+        actix_rt::spawn(async move {
+            tokio::time::delay_for(time).await;
+            actix_rt::Arbiter::current().stop();
+        });
+        let f = actix_rt::Arbiter::local_join();
+        actix_rt::spawn(async move {
+            tokio::time::delay_for(large_timer).await;
+            actix_rt::Arbiter::current().stop();
+        });
+        f.await;
+    });
+    assert!(
+        instant.elapsed() < large_timer,
+        "local_join should await only for the already spawned futures"
+    );
+}

actix-server/CHANGES.md

@@ -1,5 +1,13 @@
 # Changes
 
+## [1.0.2] - 2020-02-26
+
+### Fixed
+
+* Avoid error by calling `reregister()` on Windows [#103]
+
+[#103]: https://github.com/actix/actix-net/pull/103
+
 ## [1.0.1] - 2019-12-29
 
 ### Changed

actix-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-server"
-version = "1.0.1"
+version = "1.0.2"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix server - General purpose tcp server"
 keywords = ["network", "framework", "async", "futures"]
@@ -34,6 +34,7 @@ futures = "0.3.1"
 slab = "0.4"
 
 # unix domain sockets
+# FIXME: Remove it and use mio own uds feature once mio 0.7 is released
 mio-uds = { version = "0.6.7" }
 
 [dev-dependencies]

actix-server/src/accept.rs

@@ -298,12 +298,7 @@ impl Accept {
                     }
                     Command::Resume => {
                         for (token, info) in self.sockets.iter() {
-                            if let Err(err) = self.poll.register(
-                                &info.sock,
-                                mio::Token(token + DELTA),
-                                mio::Ready::readable(),
-                                mio::PollOpt::edge(),
-                            ) {
+                            if let Err(err) = self.register(token, info) {
                                 error!("Can not resume socket accept process: {}", err);
                             } else {
                                 info!(
@@ -338,17 +333,44 @@ impl Accept {
         true
     }
 
+    #[cfg(not(target_os = "windows"))]
+    fn register(&self, token: usize, info: &ServerSocketInfo) -> io::Result<()> {
+        self.poll.register(
+            &info.sock,
+            mio::Token(token + DELTA),
+            mio::Ready::readable(),
+            mio::PollOpt::edge(),
+        )
+    }
+
+    #[cfg(target_os = "windows")]
+    fn register(&self, token: usize, info: &ServerSocketInfo) -> io::Result<()> {
+        // On windows, calling register without deregister cause an error.
+        // See https://github.com/actix/actix-web/issues/905
+        // Calling reregister seems to fix the issue.
+        self.poll
+            .register(
+                &info.sock,
+                mio::Token(token + DELTA),
+                mio::Ready::readable(),
+                mio::PollOpt::edge(),
+            )
+            .or_else(|_| {
+                self.poll.reregister(
+                    &info.sock,
+                    mio::Token(token + DELTA),
+                    mio::Ready::readable(),
+                    mio::PollOpt::edge(),
+                )
+            })
+    }
+
     fn backpressure(&mut self, on: bool) {
         if self.backpressure {
             if !on {
                 self.backpressure = false;
                 for (token, info) in self.sockets.iter() {
-                    if let Err(err) = self.poll.register(
-                        &info.sock,
-                        mio::Token(token + DELTA),
-                        mio::Ready::readable(),
-                        mio::PollOpt::edge(),
-                    ) {
+                    if let Err(err) = self.register(token, info) {
                         error!("Can not resume socket accept process: {}", err);
                     } else {
                         info!("Accepting connections on {} has been resumed", info.addr);

actix-server/src/builder.rs

@@ -13,7 +13,6 @@ use futures::stream::FuturesUnordered;
 use futures::{ready, Future, FutureExt, Stream, StreamExt};
 use log::{error, info};
 use net2::TcpBuilder;
-use num_cpus;
 
 use crate::accept::{AcceptLoop, AcceptNotify, Command};
 use crate::config::{ConfiguredService, ServiceConfig};
@@ -220,7 +219,7 @@ impl ServerBuilder {
         self.services.push(StreamNewService::create(
             name.as_ref().to_string(),
             token,
-            factory.clone(),
+            factory,
             addr,
         ));
         self.sockets
@@ -263,12 +262,14 @@ impl ServerBuilder {
             info!("Starting {} workers", self.threads);
 
             // start workers
-            let mut workers = Vec::new();
-            for idx in 0..self.threads {
-                let worker = self.start_worker(idx, self.accept.get_notify());
-                workers.push(worker.clone());
-                self.workers.push((idx, worker));
-            }
+            let workers = (0..self.threads)
+                .map(|idx| {
+                    let worker = self.start_worker(idx, self.accept.get_notify());
+                    self.workers.push((idx, worker.clone()));
+
+                    worker
+                })
+                .collect();
 
             // start accept thread
             for sock in &self.sockets {
@@ -380,7 +381,7 @@ impl ServerBuilder {
                                             .await;
                                             System::current().stop();
                                         }
-                                        .boxed(),
+                                            .boxed(),
                                     );
                                 }
                                 ready(())

actix-server/src/config.rs

@@ -91,7 +91,7 @@ impl ConfiguredService {
 
     pub(super) fn stream(&mut self, token: Token, name: String, addr: net::SocketAddr) {
         self.names.insert(token, (name.clone(), addr));
-        self.topics.insert(name.clone(), token);
+        self.topics.insert(name, token);
         self.services.push(token);
     }
 }

actix-server/src/worker.rs

@@ -315,6 +315,8 @@ enum WorkerState {
 impl Future for Worker {
     type Output = ();
 
+    // FIXME: remove this attribute
+    #[allow(clippy::never_loop)]
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         // `StopWorker` message handler
         if let Poll::Ready(Some(StopCommand { graceful, result })) =
@@ -368,11 +370,8 @@ impl Future for Worker {
                     Ok(false) => {
                         // push connection back to queue
                         if let Some(conn) = conn {
-                            match self.state {
-                                WorkerState::Unavailable(ref mut conns) => {
-                                    conns.push(conn);
-                                }
-                                _ => (),
+                            if let WorkerState::Unavailable(ref mut conns) = self.state {
+                                conns.push(conn);
                             }
                         }
                         Poll::Pending

actix-server/tests/test_server.rs

@@ -1,15 +1,10 @@
-use std::io::Read;
 use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
 use std::sync::{mpsc, Arc};
 use std::{net, thread, time};
 
-use actix_codec::{BytesCodec, Framed};
-use actix_rt::net::TcpStream;
 use actix_server::Server;
 use actix_service::fn_service;
-use bytes::Bytes;
 use futures::future::{lazy, ok};
-use futures::SinkExt;
 use net2::TcpBuilder;
 
 fn unused_addr() -> net::SocketAddr {
@@ -41,7 +36,7 @@ fn test_bind() {
 
     thread::sleep(time::Duration::from_millis(500));
     assert!(net::TcpStream::connect(addr).is_ok());
-    let _ = sys.stop();
+    sys.stop();
     let _ = h.join();
 }
 
@@ -66,13 +61,19 @@ fn test_listen() {
 
     thread::sleep(time::Duration::from_millis(500));
     assert!(net::TcpStream::connect(addr).is_ok());
-    let _ = sys.stop();
+    sys.stop();
     let _ = h.join();
 }
 
 #[test]
 #[cfg(unix)]
 fn test_start() {
+    use actix_codec::{BytesCodec, Framed};
+    use actix_rt::net::TcpStream;
+    use bytes::Bytes;
+    use futures::SinkExt;
+    use std::io::Read;
+
     let addr = unused_addr();
     let (tx, rx) = mpsc::channel();
 
@@ -130,7 +131,7 @@ fn test_start() {
     assert!(net::TcpStream::connect(addr).is_err());
 
     thread::sleep(time::Duration::from_millis(100));
-    let _ = sys.stop();
+    sys.stop();
     let _ = h.join();
 }
 
@@ -178,6 +179,6 @@ fn test_configure() {
     assert!(net::TcpStream::connect(addr2).is_ok());
     assert!(net::TcpStream::connect(addr3).is_ok());
     assert_eq!(num.load(Relaxed), 1);
-    let _ = sys.stop();
+    sys.stop();
     let _ = h.join();
 }

actix-service/CHANGES.md

@@ -1,5 +1,30 @@
 # Changes
 
+## [1.0.5] - 2020-01-16
+
+### Fixed
+
+* Fixed unsoundness in .and_then()/.then() service combinators
+
+## [1.0.4] - 2020-01-15
+
+### Fixed
+
+* Revert 1.0.3 change
+
+## [1.0.3] - 2020-01-15
+
+### Fixed
+
+* Fixed unsoundness in `AndThenService` impl
+
+## [1.0.2] - 2020-01-08
+
+### Added
+
+* Add `into_service` helper function
+
+
 ## [1.0.1] - 2019-12-22
 
 ### Changed

actix-service/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-service"
-version = "1.0.1"
+version = "1.0.5"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix service"
 keywords = ["network", "framework", "async", "futures"]
@@ -10,12 +10,6 @@ documentation = "https://docs.rs/actix-service/"
 categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
 edition = "2018"
-workspace = ".."
-
-[badges]
-travis-ci = { repository = "actix/actix-service", branch = "master" }
-appveyor = { repository = "actix/actix-net" }
-codecov = { repository = "actix/actix-service", branch = "master", service = "github" }
 
 [lib]
 name = "actix_service"
@@ -27,3 +21,12 @@ pin-project = "0.4.6"
 
 [dev-dependencies]
 actix-rt = "1.0.0"
+criterion = "0.3"
+
+[[bench]]
+name = "unsafecell_vs_refcell"
+harness = false
+
+[[bench]]
+name = "and_then"
+harness = false

actix-service/benches/and_then.rs

@@ -0,0 +1,324 @@
+/// Benchmark various implementations of and_then
+
+use criterion::{criterion_main, Criterion};
+use futures_util::future::join_all;
+use std::cell::{RefCell, UnsafeCell};
+use std::task::{Context, Poll};
+use std::rc::Rc;
+use actix_service::{Service};
+use actix_service::IntoService;
+use std::future::Future;
+use std::pin::Pin;
+use futures_util::future::TryFutureExt;
+use actix_service::boxed::BoxFuture;
+
+
+/*
+ * Test services A,B for AndThen service implementations
+ */
+
+async fn svc1(_: ()) -> Result<usize, ()> {
+	Ok(1)
+}
+
+async fn svc2(req: usize) -> Result<usize, ()> {
+	Ok(req + 1)
+}
+
+/*
+ * AndThenUC - original AndThen service based on UnsafeCell
+ * Cut down version of actix_service::AndThenService based on actix-service::Cell
+ */
+
+
+struct AndThenUC<A,B>(Rc<UnsafeCell<(A, B)>>);
+
+impl<A,B> AndThenUC<A,B> {
+	fn new(a: A, b: B) -> Self
+	where
+		A: Service,
+		B: Service<Request = A::Response, Error = A::Error>,
+	{
+		Self(Rc::new(UnsafeCell::new((a,b))))
+	}
+}
+
+impl<A,B> Clone for AndThenUC<A,B> {
+	fn clone(&self) -> Self {
+		Self(self.0.clone())
+	}
+}
+
+impl<A,B> Service for AndThenUC<A,B>
+where
+    A: Service,
+    B: Service<Request = A::Response, Error = A::Error>
+{
+	type Request = A::Request;
+	type Response = B::Response;
+	type Error = A::Error;
+	type Future = AndThenServiceResponse<A,B>;
+
+	fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+		Poll::Ready(Ok(()))
+	}
+
+	fn call(&mut self, req: A::Request) -> Self::Future {
+		let fut = unsafe { &mut *(*self.0).get() }.0.call(req);
+		AndThenServiceResponse {
+			state: State::A(fut, Some(self.0.clone()))
+		}
+	}
+}
+
+#[pin_project::pin_project]
+pub(crate) struct AndThenServiceResponse<A, B>
+where
+    A: Service,
+    B: Service<Request = A::Response, Error = A::Error>,
+{
+    #[pin]
+    state: State<A, B>,
+}
+
+#[pin_project::pin_project]
+enum State<A, B>
+where
+    A: Service,
+    B: Service<Request = A::Response, Error = A::Error>,
+{
+    A(#[pin] A::Future, Option<Rc<UnsafeCell<(A, B)>>>),
+    B(#[pin] B::Future),
+    Empty,
+}
+
+impl<A, B> Future for AndThenServiceResponse<A, B>
+where
+    A: Service,
+    B: Service<Request = A::Response, Error = A::Error>,
+{
+    type Output = Result<B::Response, A::Error>;
+
+    #[pin_project::project]
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let mut this = self.as_mut().project();
+
+        #[project]
+        match this.state.as_mut().project() {
+            State::A(fut, b) => match fut.poll(cx)? {
+                Poll::Ready(res) => {
+                    let b = b.take().unwrap();
+                    this.state.set(State::Empty); // drop fut A
+                    let fut = unsafe { &mut (*b.get()).1 }.call(res);
+                    this.state.set(State::B(fut));
+                    self.poll(cx)
+                }
+                Poll::Pending => Poll::Pending,
+            },
+            State::B(fut) => fut.poll(cx).map(|r| {
+                this.state.set(State::Empty);
+                r
+            }),
+            State::Empty => panic!("future must not be polled after it returned `Poll::Ready`"),
+        }
+    }
+}
+
+
+/*
+ * AndThenRC - AndThen service based on RefCell
+ */
+
+ struct AndThenRC<A,B>(Rc<RefCell<(A, B)>>);
+
+ impl<A,B> AndThenRC<A,B> {
+	 fn new(a: A, b: B) -> Self
+	 where
+		 A: Service,
+		 B: Service<Request = A::Response, Error = A::Error>,
+	 {
+		 Self(Rc::new(RefCell::new((a,b))))
+	 }
+ }
+ 
+ impl<A,B> Clone for AndThenRC<A,B> {
+	 fn clone(&self) -> Self {
+		 Self(self.0.clone())
+	 }
+ }
+ 
+ impl<A,B> Service for AndThenRC<A,B>
+ where
+	 A: Service,
+	 B: Service<Request = A::Response, Error = A::Error>
+ {
+	 type Request = A::Request;
+	 type Response = B::Response;
+	 type Error = A::Error;
+	 type Future = AndThenServiceResponseRC<A,B>;
+ 
+	 fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+		 Poll::Ready(Ok(()))
+	 }
+ 
+	 fn call(&mut self, req: A::Request) -> Self::Future {
+		 let fut = self.0.borrow_mut().0.call(req);
+		 AndThenServiceResponseRC {
+			 state: StateRC::A(fut, Some(self.0.clone()))
+		 }
+	 }
+ }
+ 
+ #[pin_project::pin_project]
+ pub(crate) struct AndThenServiceResponseRC<A, B>
+ where
+	 A: Service,
+	 B: Service<Request = A::Response, Error = A::Error>,
+ {
+	 #[pin]
+	 state: StateRC<A, B>,
+ }
+ 
+ #[pin_project::pin_project]
+ enum StateRC<A, B>
+ where
+	 A: Service,
+	 B: Service<Request = A::Response, Error = A::Error>,
+ {
+	 A(#[pin] A::Future, Option<Rc<RefCell<(A, B)>>>),
+	 B(#[pin] B::Future),
+	 Empty,
+ }
+ 
+ impl<A, B> Future for AndThenServiceResponseRC<A, B>
+ where
+	 A: Service,
+	 B: Service<Request = A::Response, Error = A::Error>,
+ {
+	 type Output = Result<B::Response, A::Error>;
+ 
+	 #[pin_project::project]
+	 fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+		 let mut this = self.as_mut().project();
+ 
+		 #[project]
+		 match this.state.as_mut().project() {
+			 StateRC::A(fut, b) => match fut.poll(cx)? {
+				 Poll::Ready(res) => {
+					 let b = b.take().unwrap();
+					 this.state.set(StateRC::Empty); // drop fut A
+					 let fut = b.borrow_mut().1.call(res);
+					 this.state.set(StateRC::B(fut));
+					 self.poll(cx)
+				 }
+				 Poll::Pending => Poll::Pending,
+			 },
+			 StateRC::B(fut) => fut.poll(cx).map(|r| {
+				 this.state.set(StateRC::Empty);
+				 r
+			 }),
+			 StateRC::Empty => panic!("future must not be polled after it returned `Poll::Ready`"),
+		 }
+	 }
+ }
+ 
+
+ /*
+ * AndThenRCFuture - AndThen service based on RefCell 
+ * and standard futures::future::and_then combinator in a Box
+ */
+
+ struct AndThenRCFuture<A,B>(Rc<RefCell<(A, B)>>);
+
+ impl<A,B> AndThenRCFuture<A,B> {
+	 fn new(a: A, b: B) -> Self
+	 where
+		 A: Service,
+		 B: Service<Request = A::Response, Error = A::Error>,
+	 {
+		 Self(Rc::new(RefCell::new((a,b))))
+	 }
+ }
+ 
+ impl<A,B> Clone for AndThenRCFuture<A,B> {
+	 fn clone(&self) -> Self {
+		 Self(self.0.clone())
+	 }
+ }
+ 
+ impl<A,B> Service for AndThenRCFuture<A,B>
+ where
+	A: Service + 'static,
+	A::Future: 'static,
+	B: Service<Request = A::Response, Error = A::Error> + 'static,
+ 	B::Future: 'static
+ {
+	 type Request = A::Request;
+	 type Response = B::Response;
+	 type Error = A::Error;
+	 type Future = BoxFuture<Self::Response, Self::Error>;
+ 
+	 fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+		 Poll::Ready(Ok(()))
+	 }
+ 
+	 fn call(&mut self, req: A::Request) -> Self::Future {
+		 let fut = self.0.borrow_mut().0.call(req);
+		 let core = self.0.clone();
+		 let fut2 = move |res| (*core).borrow_mut().1.call(res);
+		 Box::pin(
+			fut.and_then(fut2)
+		)
+	 }
+ }
+ 
+
+/// Criterion Benchmark for async Service
+/// Should be used from within criterion group:
+/// ```rust,ignore
+/// let mut criterion: ::criterion::Criterion<_> =
+///     ::criterion::Criterion::default().configure_from_args();
+/// bench_async_service(&mut criterion, ok_service(), "async_service_direct");
+/// ```
+///
+/// Usable for benching Service wrappers:
+/// Using minimum service code implementation we first measure
+/// time to run minimum service, then measure time with wrapper.
+///
+/// Sample output
+/// async_service_direct    time:   [1.0908 us 1.1656 us 1.2613 us]
+pub fn bench_async_service<S>(c: &mut Criterion, srv: S, name: &str)
+where
+    S: Service<Request = (), Response = usize, Error = ()> + Clone + 'static,
+{
+    let mut rt = actix_rt::System::new("test");
+
+    // start benchmark loops
+    c.bench_function(name, move |b| {
+        b.iter_custom(|iters| {
+            let mut srvs: Vec<_> = (1..iters).map(|_| srv.clone()).collect();
+            // exclude request generation, it appears it takes significant time vs call (3us vs 1us)
+            let start = std::time::Instant::now();
+            // benchmark body
+            rt.block_on(async move {
+				join_all(srvs.iter_mut().map(|srv| srv.call(()))).await
+			});
+            let elapsed = start.elapsed();
+            // check that at least first request succeeded
+            elapsed
+        })
+    });
+}
+
+
+pub fn service_benches() {
+    let mut criterion: ::criterion::Criterion<_> =
+        ::criterion::Criterion::default().configure_from_args();
+    bench_async_service(&mut criterion, AndThenUC::new(svc1.into_service(), svc2.into_service()), "AndThen with UnsafeCell");
+    bench_async_service(&mut criterion, AndThenRC::new(svc1.into_service(), svc2.into_service()), "AndThen with RefCell");
+    bench_async_service(&mut criterion, AndThenUC::new(svc1.into_service(), svc2.into_service()), "AndThen with UnsafeCell");
+    bench_async_service(&mut criterion, AndThenRC::new(svc1.into_service(), svc2.into_service()), "AndThen with RefCell");
+    bench_async_service(&mut criterion, AndThenRCFuture::new(svc1.into_service(), svc2.into_service()), "AndThen with RefCell via future::and_then");
+}
+
+criterion_main!(service_benches);

actix-service/benches/unsafecell_vs_refcell.rs

@@ -0,0 +1,117 @@
+use criterion::{criterion_main, Criterion};
+use futures_util::future::join_all;
+use std::cell::{RefCell, UnsafeCell};
+use std::task::{Context, Poll};
+use std::rc::Rc;
+use actix_service::{Service};
+use futures_util::future::{ok, Ready};
+
+struct SrvUC(Rc<UnsafeCell<usize>>);
+
+impl Default for SrvUC {
+	fn default() -> Self {
+		Self(Rc::new(UnsafeCell::new(0)))
+	}
+}
+
+impl Clone for SrvUC {
+	fn clone(&self) -> Self {
+		Self(self.0.clone())
+	}
+}
+
+impl Service for SrvUC {
+	type Request = ();
+	type Response = usize;
+	type Error = ();
+	type Future = Ready<Result<Self::Response, ()>>;
+
+	fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+		Poll::Ready(Ok(()))
+	}
+
+	fn call(&mut self, _: ()) -> Self::Future {
+		unsafe { *(*self.0).get() = *(*self.0).get() + 1 };
+		ok(unsafe { *self.0.get() })
+	}
+}
+
+struct SrvRC(Rc<RefCell<usize>>);
+
+impl Default for SrvRC {
+	fn default() -> Self {
+		Self(Rc::new(RefCell::new(0)))
+	}
+}
+
+impl Clone for SrvRC {
+	fn clone(&self) -> Self {
+		Self(self.0.clone())
+	}
+}
+
+impl Service for SrvRC {
+	type Request = ();
+	type Response = usize;
+	type Error = ();
+	type Future = Ready<Result<Self::Response, ()>>;
+
+	fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+		Poll::Ready(Ok(()))
+	}
+
+	fn call(&mut self, _: ()) -> Self::Future {
+		let prev = *self.0.borrow();
+		*(*self.0).borrow_mut() = prev + 1;
+		ok(*self.0.borrow())
+	}
+}
+
+
+/// Criterion Benchmark for async Service
+/// Should be used from within criterion group:
+/// ```rust,ignore
+/// let mut criterion: ::criterion::Criterion<_> =
+///     ::criterion::Criterion::default().configure_from_args();
+/// bench_async_service(&mut criterion, ok_service(), "async_service_direct");
+/// ```
+///
+/// Usable for benching Service wrappers:
+/// Using minimum service code implementation we first measure
+/// time to run minimum service, then measure time with wrapper.
+///
+/// Sample output
+/// async_service_direct    time:   [1.0908 us 1.1656 us 1.2613 us]
+pub fn bench_async_service<S>(c: &mut Criterion, srv: S, name: &str)
+where
+    S: Service<Request = (), Response = usize, Error = ()> + Clone + 'static,
+{
+    let mut rt = actix_rt::System::new("test");
+
+    // start benchmark loops
+    c.bench_function(name, move |b| {
+        b.iter_custom(|iters| {
+            let mut srvs: Vec<_> = (1..iters).map(|_| srv.clone()).collect();
+            // exclude request generation, it appears it takes significant time vs call (3us vs 1us)
+            let start = std::time::Instant::now();
+            // benchmark body
+            rt.block_on(async move {
+				join_all(srvs.iter_mut().map(|srv| srv.call(()))).await
+			});
+            let elapsed = start.elapsed();
+            // check that at least first request succeeded
+            elapsed
+        })
+    });
+}
+
+
+pub fn service_benches() {
+    let mut criterion: ::criterion::Criterion<_> =
+        ::criterion::Criterion::default().configure_from_args();
+    bench_async_service(&mut criterion, SrvUC::default(), "Service with UnsafeCell");
+    bench_async_service(&mut criterion, SrvRC::default(), "Service with RefCell");
+    bench_async_service(&mut criterion, SrvUC::default(), "Service with UnsafeCell");
+    bench_async_service(&mut criterion, SrvRC::default(), "Service with RefCell");
+}
+criterion_main!(service_benches);

actix-service/src/and_then.rs

@@ -1,5 +1,6 @@
 use std::future::Future;
 use std::pin::Pin;
+use std::rc::Rc;
 use std::task::{Context, Poll};
 
 use super::{Service, ServiceFactory};
@@ -9,7 +10,7 @@ use crate::cell::Cell;
 /// of another service which completes successfully.
 ///
 /// This is created by the `ServiceExt::and_then` method.
-pub struct AndThenService<A, B>(Cell<(A, B)>);
+pub(crate) struct AndThenService<A, B>(Cell<(A, B)>);
 
 impl<A, B> AndThenService<A, B> {
     /// Create new `AndThen` combinator
@@ -40,7 +41,6 @@ where
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         let srv = self.0.get_mut();
-
         let not_ready = !srv.0.poll_ready(cx)?.is_ready();
         if !srv.1.poll_ready(cx)?.is_ready() || not_ready {
             Poll::Pending
@@ -57,7 +57,7 @@ where
 }
 
 #[pin_project::pin_project]
-pub struct AndThenServiceResponse<A, B>
+pub(crate) struct AndThenServiceResponse<A, B>
 where
     A: Service,
     B: Service<Request = A::Response, Error = A::Error>,
@@ -110,7 +110,7 @@ where
 }
 
 /// `.and_then()` service factory combinator
-pub struct AndThenServiceFactory<A, B>
+pub(crate) struct AndThenServiceFactory<A, B>
 where
     A: ServiceFactory,
     A::Config: Clone,
@@ -121,8 +121,7 @@ where
         InitError = A::InitError,
     >,
 {
-    a: A,
-    b: B,
+    inner: Rc<(A, B)>,
 }
 
 impl<A, B> AndThenServiceFactory<A, B>
@@ -138,7 +137,9 @@ where
 {
     /// Create new `AndThenFactory` combinator
     pub(crate) fn new(a: A, b: B) -> Self {
-        Self { a, b }
+        Self {
+            inner: Rc::new((a, b)),
+        }
     }
 }
 
@@ -163,34 +164,34 @@ where
     type Future = AndThenServiceFactoryResponse<A, B>;
 
     fn new_service(&self, cfg: A::Config) -> Self::Future {
+        let inner = &*self.inner;
         AndThenServiceFactoryResponse::new(
-            self.a.new_service(cfg.clone()),
-            self.b.new_service(cfg),
+            inner.0.new_service(cfg.clone()),
+            inner.1.new_service(cfg),
         )
     }
 }
 
 impl<A, B> Clone for AndThenServiceFactory<A, B>
 where
-    A: ServiceFactory + Clone,
+    A: ServiceFactory,
     A::Config: Clone,
     B: ServiceFactory<
-            Config = A::Config,
-            Request = A::Response,
-            Error = A::Error,
-            InitError = A::InitError,
-        > + Clone,
+        Config = A::Config,
+        Request = A::Response,
+        Error = A::Error,
+        InitError = A::InitError,
+    >,
 {
     fn clone(&self) -> Self {
         Self {
-            a: self.a.clone(),
-            b: self.b.clone(),
+            inner: self.inner.clone(),
         }
     }
 }
 
 #[pin_project::pin_project]
-pub struct AndThenServiceFactoryResponse<A, B>
+pub(crate) struct AndThenServiceFactoryResponse<A, B>
 where
     A: ServiceFactory,
     B: ServiceFactory<Request = A::Response>,
@@ -312,7 +313,7 @@ mod tests {
         let mut srv = pipeline(Srv1(cnt.clone())).and_then(Srv2(cnt));
         let res = srv.call("srv1").await;
         assert!(res.is_ok());
-        assert_eq!(res.unwrap(), (("srv1", "srv2")));
+        assert_eq!(res.unwrap(), ("srv1", "srv2"));
     }
 
     #[actix_rt::test]

actix-service/src/and_then_apply_fn.rs

@@ -1,13 +1,14 @@
 use std::future::Future;
 use std::marker::PhantomData;
 use std::pin::Pin;
+use std::rc::Rc;
 use std::task::{Context, Poll};
 
 use crate::cell::Cell;
 use crate::{Service, ServiceFactory};
 
 /// `Apply` service combinator
-pub struct AndThenApplyFn<A, B, F, Fut, Res, Err>
+pub(crate) struct AndThenApplyFn<A, B, F, Fut, Res, Err>
 where
     A: Service,
     B: Service,
@@ -15,8 +16,7 @@ where
     Fut: Future<Output = Result<Res, Err>>,
     Err: From<A::Error> + From<B::Error>,
 {
-    a: A,
-    b: Cell<(B, F)>,
+    srv: Cell<(A, B, F)>,
     r: PhantomData<(Fut, Res, Err)>,
 }
 
@@ -31,8 +31,7 @@ where
     /// Create new `Apply` combinator
     pub(crate) fn new(a: A, b: B, f: F) -> Self {
         Self {
-            a,
-            b: Cell::new((b, f)),
+            srv: Cell::new((a, b, f)),
             r: PhantomData,
         }
     }
@@ -40,7 +39,7 @@ where
 
 impl<A, B, F, Fut, Res, Err> Clone for AndThenApplyFn<A, B, F, Fut, Res, Err>
 where
-    A: Service + Clone,
+    A: Service,
     B: Service,
     F: FnMut(A::Response, &mut B) -> Fut,
     Fut: Future<Output = Result<Res, Err>>,
@@ -48,8 +47,7 @@ where
 {
     fn clone(&self) -> Self {
         AndThenApplyFn {
-            a: self.a.clone(),
-            b: self.b.clone(),
+            srv: self.srv.clone(),
             r: PhantomData,
         }
     }
@@ -69,8 +67,9 @@ where
     type Future = AndThenApplyFnFuture<A, B, F, Fut, Res, Err>;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        let not_ready = self.a.poll_ready(cx)?.is_pending();
-        if self.b.get_mut().0.poll_ready(cx)?.is_pending() || not_ready {
+        let inner = self.srv.get_mut();
+        let not_ready = inner.0.poll_ready(cx)?.is_pending();
+        if inner.1.poll_ready(cx)?.is_pending() || not_ready {
             Poll::Pending
         } else {
             Poll::Ready(Ok(()))
@@ -78,14 +77,15 @@ where
     }
 
     fn call(&mut self, req: A::Request) -> Self::Future {
+        let fut = self.srv.get_mut().0.call(req);
         AndThenApplyFnFuture {
-            state: State::A(self.a.call(req), Some(self.b.clone())),
+            state: State::A(fut, Some(self.srv.clone())),
         }
     }
 }
 
 #[pin_project::pin_project]
-pub struct AndThenApplyFnFuture<A, B, F, Fut, Res, Err>
+pub(crate) struct AndThenApplyFnFuture<A, B, F, Fut, Res, Err>
 where
     A: Service,
     B: Service,
@@ -108,7 +108,7 @@ where
     Err: From<A::Error>,
     Err: From<B::Error>,
 {
-    A(#[pin] A::Future, Option<Cell<(B, F)>>),
+    A(#[pin] A::Future, Option<Cell<(A, B, F)>>),
     B(#[pin] Fut),
     Empty,
 }
@@ -134,7 +134,7 @@ where
                     let mut b = b.take().unwrap();
                     this.state.set(State::Empty);
                     let b = b.get_mut();
-                    let fut = (&mut b.1)(res, &mut b.0);
+                    let fut = (&mut b.2)(res, &mut b.1);
                     this.state.set(State::B(fut));
                     self.poll(cx)
                 }
@@ -150,10 +150,8 @@ where
 }
 
 /// `AndThenApplyFn` service factory
-pub struct AndThenApplyFnFactory<A, B, F, Fut, Res, Err> {
-    a: A,
-    b: B,
-    f: F,
+pub(crate) struct AndThenApplyFnFactory<A, B, F, Fut, Res, Err> {
+    srv: Rc<(A, B, F)>,
     r: PhantomData<(Fut, Res, Err)>,
 }
 
@@ -168,25 +166,16 @@ where
     /// Create new `ApplyNewService` new service instance
     pub(crate) fn new(a: A, b: B, f: F) -> Self {
         Self {
-            a: a,
-            b: b,
-            f: f,
+            srv: Rc::new((a, b, f)),
             r: PhantomData,
         }
     }
 }
 
-impl<A, B, F, Fut, Res, Err> Clone for AndThenApplyFnFactory<A, B, F, Fut, Res, Err>
-where
-    A: Clone,
-    B: Clone,
-    F: Clone,
-{
+impl<A, B, F, Fut, Res, Err> Clone for AndThenApplyFnFactory<A, B, F, Fut, Res, Err> {
     fn clone(&self) -> Self {
         Self {
-            a: self.a.clone(),
-            b: self.b.clone(),
-            f: self.f.clone(),
+            srv: self.srv.clone(),
             r: PhantomData,
         }
     }
@@ -210,18 +199,19 @@ where
     type Future = AndThenApplyFnFactoryResponse<A, B, F, Fut, Res, Err>;
 
     fn new_service(&self, cfg: A::Config) -> Self::Future {
+        let srv = &*self.srv;
         AndThenApplyFnFactoryResponse {
             a: None,
             b: None,
-            f: self.f.clone(),
-            fut_a: self.a.new_service(cfg.clone()),
-            fut_b: self.b.new_service(cfg),
+            f: srv.2.clone(),
+            fut_a: srv.0.new_service(cfg.clone()),
+            fut_b: srv.1.new_service(cfg),
         }
     }
 }
 
 #[pin_project::pin_project]
-pub struct AndThenApplyFnFactoryResponse<A, B, F, Fut, Res, Err>
+pub(crate) struct AndThenApplyFnFactoryResponse<A, B, F, Fut, Res, Err>
 where
     A: ServiceFactory,
     B: ServiceFactory<Config = A::Config, InitError = A::InitError>,
@@ -267,8 +257,11 @@ where
 
         if this.a.is_some() && this.b.is_some() {
             Poll::Ready(Ok(AndThenApplyFn {
-                a: this.a.take().unwrap(),
-                b: Cell::new((this.b.take().unwrap(), this.f.clone())),
+                srv: Cell::new((
+                    this.a.take().unwrap(),
+                    this.b.take().unwrap(),
+                    this.f.clone(),
+                )),
                 r: PhantomData,
             }))
         } else {
@@ -297,6 +290,7 @@ mod tests {
             Poll::Ready(Ok(()))
         }
 
+        #[allow(clippy::unit_arg)]
         fn call(&mut self, req: Self::Request) -> Self::Future {
             ok(req)
         }
@@ -304,7 +298,7 @@ mod tests {
 
     #[actix_rt::test]
     async fn test_service() {
-        let mut srv = pipeline(|r: &'static str| ok(r))
+        let mut srv = pipeline(ok)
             .and_then_apply_fn(Srv, |req: &'static str, s| {
                 s.call(()).map_ok(move |res| (req, res))
             });
@@ -318,7 +312,7 @@ mod tests {
 
     #[actix_rt::test]
     async fn test_service_factory() {
-        let new_srv = pipeline_factory(|| ok::<_, ()>(fn_service(|r: &'static str| ok(r))))
+        let new_srv = pipeline_factory(|| ok::<_, ()>(fn_service(ok)))
             .and_then_apply_fn(
                 || ok(Srv),
                 |req: &'static str, s| s.call(()).map_ok(move |res| (req, res)),

actix-service/src/apply.rs

@@ -233,8 +233,8 @@ mod tests {
         let mut srv = pipeline(apply_fn(Srv, |req: &'static str, srv| {
             let fut = srv.call(());
             async move {
-                let res = fut.await.unwrap();
-                Ok((req, res))
+                fut.await.unwrap();
+                Ok((req, ()))
             }
         }));
 
@@ -242,7 +242,7 @@ mod tests {
 
         let res = srv.call("srv").await;
         assert!(res.is_ok());
-        assert_eq!(res.unwrap(), (("srv", ())));
+        assert_eq!(res.unwrap(), ("srv", ()));
     }
 
     #[actix_rt::test]
@@ -252,8 +252,8 @@ mod tests {
             |req: &'static str, srv| {
                 let fut = srv.call(());
                 async move {
-                    let res = fut.await.unwrap();
-                    Ok((req, res))
+                    fut.await.unwrap();
+                    Ok((req, ()))
                 }
             },
         ));
@@ -264,6 +264,6 @@ mod tests {
 
         let res = srv.call("srv").await;
         assert!(res.is_ok());
-        assert_eq!(res.unwrap(), (("srv", ())));
+        assert_eq!(res.unwrap(), ("srv", ()));
     }
 }

actix-service/src/apply_cfg.rs

@@ -7,7 +7,18 @@ use crate::cell::Cell;
 use crate::{Service, ServiceFactory};
 
 /// Convert `Fn(Config, &mut Service1) -> Future<Service2>` fn to a service factory
-pub fn apply_cfg<F, C, T, R, S, E>(srv: T, f: F) -> ApplyConfigService<F, C, T, R, S, E>
+pub fn apply_cfg<F, C, T, R, S, E>(
+    srv: T,
+    f: F,
+) -> impl ServiceFactory<
+    Config = C,
+    Request = S::Request,
+    Response = S::Response,
+    Error = S::Error,
+    Service = S,
+    InitError = E,
+    Future = R,
+> + Clone
 where
     F: FnMut(C, &mut T) -> R,
     T: Service,
@@ -26,7 +37,14 @@ where
 pub fn apply_cfg_factory<F, C, T, R, S>(
     factory: T,
     f: F,
-) -> ApplyConfigServiceFactory<F, C, T, R, S>
+) -> impl ServiceFactory<
+    Config = C,
+    Request = S::Request,
+    Response = S::Response,
+    Error = S::Error,
+    Service = S,
+    InitError = T::InitError,
+> + Clone
 where
     F: FnMut(C, &mut T::Service) -> R,
     T: ServiceFactory<Config = ()>,
@@ -41,7 +59,7 @@ where
 }
 
 /// Convert `Fn(Config, &mut Server) -> Future<Service>` fn to NewService\
-pub struct ApplyConfigService<F, C, T, R, S, E>
+struct ApplyConfigService<F, C, T, R, S, E>
 where
     F: FnMut(C, &mut T) -> R,
     T: Service,
@@ -92,7 +110,7 @@ where
 }
 
 /// Convert `Fn(&Config) -> Future<Service>` fn to NewService
-pub struct ApplyConfigServiceFactory<F, C, T, R, S>
+struct ApplyConfigServiceFactory<F, C, T, R, S>
 where
     F: FnMut(C, &mut T::Service) -> R,
     T: ServiceFactory<Config = ()>,
@@ -145,7 +163,7 @@ where
 }
 
 #[pin_project::pin_project]
-pub struct ApplyConfigServiceFactoryResponse<F, C, T, R, S>
+struct ApplyConfigServiceFactoryResponse<F, C, T, R, S>
 where
     F: FnMut(C, &mut T::Service) -> R,
     T: ServiceFactory<Config = ()>,

actix-service/src/lib.rs

@@ -57,7 +57,7 @@ pub use self::transform::{apply, Transform};
 ///
 ///      fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { ... }
 ///
-///      fn call(&mut self) -> Self::Future { ... }
+///      fn call(&mut self, req: Self::Request) -> Self::Future { ... }
 /// }
 /// ```
 ///
@@ -82,7 +82,7 @@ pub trait Service {
 
     /// Returns `Ready` when the service is able to process requests.
     ///
-    /// If the service is at capacity, then `NotReady` is returned and the task
+    /// If the service is at capacity, then `Pending` is returned and the task
     /// is notified when the service becomes ready again. This function is
     /// expected to be called while on a task.
     ///
@@ -351,11 +351,17 @@ where
     }
 }
 
+/// Convert object of type `T` to a service `S`
+pub fn into_service<T, S>(tp: T) -> S
+where
+    S: Service,
+    T: IntoService<S>,
+{
+    tp.into_service()
+}
+
 pub mod dev {
-    pub use crate::and_then::{AndThenService, AndThenServiceFactory};
-    pub use crate::and_then_apply_fn::{AndThenApplyFn, AndThenApplyFnFactory};
     pub use crate::apply::{Apply, ApplyServiceFactory};
-    pub use crate::apply_cfg::{ApplyConfigService, ApplyConfigServiceFactory};
     pub use crate::fn_service::{
         FnService, FnServiceConfig, FnServiceFactory, FnServiceNoConfig,
     };
@@ -363,7 +369,6 @@ pub mod dev {
     pub use crate::map_config::{MapConfig, UnitConfig};
     pub use crate::map_err::{MapErr, MapErrServiceFactory};
     pub use crate::map_init_err::MapInitErr;
-    pub use crate::then::{ThenService, ThenServiceFactory};
     pub use crate::transform::ApplyTransform;
     pub use crate::transform_err::TransformMapInitErr;
 }

actix-service/src/pipeline.rs

@@ -46,7 +46,12 @@ impl<T: Service> Pipeline<T> {
     ///
     /// Note that this function consumes the receiving service and returns a
     /// wrapped version of it.
-    pub fn and_then<F, U>(self, service: F) -> Pipeline<AndThenService<T, U>>
+    pub fn and_then<F, U>(
+        self,
+        service: F,
+    ) -> Pipeline<
+        impl Service<Request = T::Request, Response = U::Response, Error = T::Error> + Clone,
+    >
     where
         Self: Sized,
         F: IntoService<U>,
@@ -65,7 +70,7 @@ impl<T: Service> Pipeline<T> {
         self,
         service: I,
         f: F,
-    ) -> Pipeline<AndThenApplyFn<T, U, F, Fut, Res, Err>>
+    ) -> Pipeline<impl Service<Request = T::Request, Response = Res, Error = Err> + Clone>
     where
         Self: Sized,
         I: IntoService<U>,
@@ -84,7 +89,12 @@ impl<T: Service> Pipeline<T> {
     ///
     /// Note that this function consumes the receiving pipeline and returns a
     /// wrapped version of it.
-    pub fn then<F, U>(self, service: F) -> Pipeline<ThenService<T, U>>
+    pub fn then<F, U>(
+        self,
+        service: F,
+    ) -> Pipeline<
+        impl Service<Request = T::Request, Response = U::Response, Error = T::Error> + Clone,
+    >
     where
         Self: Sized,
         F: IntoService<U>,
@@ -168,7 +178,23 @@ pub struct PipelineFactory<T> {
 
 impl<T: ServiceFactory> PipelineFactory<T> {
     /// Call another service after call to this one has resolved successfully.
-    pub fn and_then<F, U>(self, factory: F) -> PipelineFactory<AndThenServiceFactory<T, U>>
+    pub fn and_then<F, U>(
+        self,
+        factory: F,
+    ) -> PipelineFactory<
+        impl ServiceFactory<
+                Request = T::Request,
+                Response = U::Response,
+                Error = T::Error,
+                Config = T::Config,
+                InitError = T::InitError,
+                Service = impl Service<
+                    Request = T::Request,
+                    Response = U::Response,
+                    Error = T::Error,
+                > + Clone,
+            > + Clone,
+    >
     where
         Self: Sized,
         T::Config: Clone,
@@ -193,7 +219,16 @@ impl<T: ServiceFactory> PipelineFactory<T> {
         self,
         factory: I,
         f: F,
-    ) -> PipelineFactory<AndThenApplyFnFactory<T, U, F, Fut, Res, Err>>
+    ) -> PipelineFactory<
+        impl ServiceFactory<
+                Request = T::Request,
+                Response = Res,
+                Error = Err,
+                Config = T::Config,
+                InitError = T::InitError,
+                Service = impl Service<Request = T::Request, Response = Res, Error = Err> + Clone,
+            > + Clone,
+    >
     where
         Self: Sized,
         T::Config: Clone,
@@ -214,7 +249,23 @@ impl<T: ServiceFactory> PipelineFactory<T> {
     ///
     /// Note that this function consumes the receiving pipeline and returns a
     /// wrapped version of it.
-    pub fn then<F, U>(self, factory: F) -> PipelineFactory<ThenServiceFactory<T, U>>
+    pub fn then<F, U>(
+        self,
+        factory: F,
+    ) -> PipelineFactory<
+        impl ServiceFactory<
+                Request = T::Request,
+                Response = U::Response,
+                Error = T::Error,
+                Config = T::Config,
+                InitError = T::InitError,
+                Service = impl Service<
+                    Request = T::Request,
+                    Response = U::Response,
+                    Error = T::Error,
+                > + Clone,
+            > + Clone,
+    >
     where
         Self: Sized,
         T::Config: Clone,

actix-service/src/then.rs

@@ -1,5 +1,6 @@
 use std::future::Future;
 use std::pin::Pin;
+use std::rc::Rc;
 use std::task::{Context, Poll};
 
 use super::{Service, ServiceFactory};
@@ -8,11 +9,8 @@ use crate::cell::Cell;
 /// Service for the `then` combinator, chaining a computation onto the end of
 /// another service.
 ///
-/// This is created by the `ServiceExt::then` method.
-pub struct ThenService<A, B> {
-    a: A,
-    b: Cell<B>,
-}
+/// This is created by the `Pipeline::then` method.
+pub(crate) struct ThenService<A, B>(Cell<(A, B)>);
 
 impl<A, B> ThenService<A, B> {
     /// Create new `.then()` combinator
@@ -21,19 +19,13 @@ impl<A, B> ThenService<A, B> {
         A: Service,
         B: Service<Request = Result<A::Response, A::Error>, Error = A::Error>,
     {
-        Self { a, b: Cell::new(b) }
+        Self(Cell::new((a, b)))
     }
 }
 
-impl<A, B> Clone for ThenService<A, B>
-where
-    A: Clone,
-{
+impl<A, B> Clone for ThenService<A, B> {
     fn clone(&self) -> Self {
-        ThenService {
-            a: self.a.clone(),
-            b: self.b.clone(),
-        }
+        ThenService(self.0.clone())
     }
 }
 
@@ -47,9 +39,10 @@ where
     type Error = B::Error;
     type Future = ThenServiceResponse<A, B>;
 
-    fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        let not_ready = !self.a.poll_ready(ctx)?.is_ready();
-        if !self.b.get_mut().poll_ready(ctx)?.is_ready() || not_ready {
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        let srv = self.0.get_mut();
+        let not_ready = !srv.0.poll_ready(cx)?.is_ready();
+        if !srv.1.poll_ready(cx)?.is_ready() || not_ready {
             Poll::Pending
         } else {
             Poll::Ready(Ok(()))
@@ -58,13 +51,13 @@ where
 
     fn call(&mut self, req: A::Request) -> Self::Future {
         ThenServiceResponse {
-            state: State::A(self.a.call(req), Some(self.b.clone())),
+            state: State::A(self.0.get_mut().0.call(req), Some(self.0.clone())),
         }
     }
 }
 
 #[pin_project::pin_project]
-pub struct ThenServiceResponse<A, B>
+pub(crate) struct ThenServiceResponse<A, B>
 where
     A: Service,
     B: Service<Request = Result<A::Response, A::Error>>,
@@ -79,7 +72,7 @@ where
     A: Service,
     B: Service<Request = Result<A::Response, A::Error>>,
 {
-    A(#[pin] A::Future, Option<Cell<B>>),
+    A(#[pin] A::Future, Option<Cell<(A, B)>>),
     B(#[pin] B::Future),
     Empty,
 }
@@ -101,7 +94,7 @@ where
                 Poll::Ready(res) => {
                     let mut b = b.take().unwrap();
                     this.state.set(State::Empty); // drop fut A
-                    let fut = b.get_mut().call(res);
+                    let fut = b.get_mut().1.call(res);
                     this.state.set(State::B(fut));
                     self.poll(cx)
                 }
@@ -117,10 +110,7 @@ where
 }
 
 /// `.then()` service factory combinator
-pub struct ThenServiceFactory<A, B> {
-    a: A,
-    b: B,
-}
+pub(crate) struct ThenServiceFactory<A, B>(Rc<(A, B)>);
 
 impl<A, B> ThenServiceFactory<A, B>
 where
@@ -135,7 +125,7 @@ where
 {
     /// Create new `AndThen` combinator
     pub(crate) fn new(a: A, b: B) -> Self {
-        Self { a, b }
+        Self(Rc::new((a, b)))
     }
 }
 
@@ -160,28 +150,19 @@ where
     type Future = ThenServiceFactoryResponse<A, B>;
 
     fn new_service(&self, cfg: A::Config) -> Self::Future {
-        ThenServiceFactoryResponse::new(
-            self.a.new_service(cfg.clone()),
-            self.b.new_service(cfg),
-        )
+        let srv = &*self.0;
+        ThenServiceFactoryResponse::new(srv.0.new_service(cfg.clone()), srv.1.new_service(cfg))
     }
 }
 
-impl<A, B> Clone for ThenServiceFactory<A, B>
-where
-    A: Clone,
-    B: Clone,
-{
+impl<A, B> Clone for ThenServiceFactory<A, B> {
     fn clone(&self) -> Self {
-        Self {
-            a: self.a.clone(),
-            b: self.b.clone(),
-        }
+        Self(self.0.clone())
     }
 }
 
 #[pin_project::pin_project]
-pub struct ThenServiceFactoryResponse<A, B>
+pub(crate) struct ThenServiceFactoryResponse<A, B>
 where
     A: ServiceFactory,
     B: ServiceFactory<
@@ -324,11 +305,11 @@ mod tests {
 
         let res = srv.call(Ok("srv1")).await;
         assert!(res.is_ok());
-        assert_eq!(res.unwrap(), (("srv1", "ok")));
+        assert_eq!(res.unwrap(), ("srv1", "ok"));
 
         let res = srv.call(Err("srv")).await;
         assert!(res.is_ok());
-        assert_eq!(res.unwrap(), (("srv2", "err")));
+        assert_eq!(res.unwrap(), ("srv2", "err"));
     }
 
     #[actix_rt::test]
@@ -340,10 +321,10 @@ mod tests {
         let mut srv = factory.new_service(&()).await.unwrap();
         let res = srv.call(Ok("srv1")).await;
         assert!(res.is_ok());
-        assert_eq!(res.unwrap(), (("srv1", "ok")));
+        assert_eq!(res.unwrap(), ("srv1", "ok"));
 
         let res = srv.call(Err("srv")).await;
         assert!(res.is_ok());
-        assert_eq!(res.unwrap(), (("srv2", "err")));
+        assert_eq!(res.unwrap(), ("srv2", "err"));
     }
 }

actix-testing/src/lib.rs

@@ -1,6 +1,6 @@
 //! Various helpers for Actix applications to use during testing.
 #![deny(rust_2018_idioms, warnings)]
-#![allow(clippy::type_complexity)]
+#![allow(clippy::type_complexity, clippy::needless_doctest_main)]
 
 use std::sync::mpsc;
 use std::{net, thread};

actix-tls/CHANGES.md

@@ -1,5 +1,13 @@
 # Changes
 
+## [2.0.0-alpha.1] - 2020-03-03
+
+### Changed
+
+* Update `rustls` dependency to 0.17
+* Update `tokio-rustls` dependency to 0.13
+* Update `webpki-roots` dependency to 0.19
+
 ## [1.0.0] - 2019-12-11
 
 * 1.0.0 release

actix-tls/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-tls"
-version = "1.0.0"
+version = "2.0.0-alpha.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix tls services"
 keywords = ["network", "framework", "async", "futures"]
@@ -46,10 +46,10 @@ open-ssl = { version="0.10", package = "openssl", optional = true }
 tokio-openssl = { version = "0.4.0", optional = true }
 
 # rustls
-rust-tls = { version = "0.16.0", package = "rustls", optional = true }
+rust-tls = { version = "0.17.0", package = "rustls", optional = true }
 webpki = { version = "0.21", optional = true }
-webpki-roots = { version = "0.17", optional = true }
-tokio-rustls = { version = "0.12.0", optional = true }
+webpki-roots = { version = "0.19", optional = true }
+tokio-rustls = { version = "0.13.0", optional = true }
 
 # native-tls
 native-tls = { version="0.2", optional = true }

actix-tracing/CHANGES.md

@@ -0,0 +1,5 @@
+# Changes
+
+## [0.1.0] - 2020-01-15
+
+* Initial release

actix-tracing/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "actix-tracing"
+version = "0.1.0"
+authors = ["Rajasekharan Vengalil <avranju@gmail.com>"]
+description = "Support for tokio tracing with Actix services"
+keywords = ["network", "framework", "tracing"]
+homepage = "https://actix.rs"
+repository = "https://github.com/actix/actix-net.git"
+documentation = "https://docs.rs/actix-tracing/"
+categories = ["network-programming", "asynchronous"]
+license = "MIT/Apache-2.0"
+edition = "2018"
+
+[lib]
+name = "actix_tracing"
+path = "src/lib.rs"
+
+[dependencies]
+actix-service = "1.0.4"
+futures-util = "0.3.1"
+tracing = "0.1"
+tracing-futures = "0.2"
+
+[dev_dependencies]
+actix-rt = "1.0"
+slab = "0.4"

actix-tracing/LICENSE-APACHE

@@ -0,0 +1 @@
+../LICENSE-APACHE

actix-tracing/LICENSE-MIT

@@ -0,0 +1 @@
+../LICENSE-MIT

actix-tracing/src/lib.rs

@@ -0,0 +1,261 @@
+//! Actix tracing - support for tokio tracing with Actix services.
+#![deny(rust_2018_idioms, warnings)]
+
+use std::marker::PhantomData;
+use std::task::{Context, Poll};
+
+use actix_service::{
+    apply, dev::ApplyTransform, IntoServiceFactory, Service, ServiceFactory, Transform,
+};
+use futures_util::future::{ok, Either, Ready};
+use tracing_futures::{Instrument, Instrumented};
+
+/// A `Service` implementation that automatically enters/exits tracing spans
+/// for the wrapped inner service.
+#[derive(Clone)]
+pub struct TracingService<S, F> {
+    inner: S,
+    make_span: F,
+}
+
+impl<S, F> TracingService<S, F> {
+    pub fn new(inner: S, make_span: F) -> Self {
+        TracingService { inner, make_span }
+    }
+}
+
+impl<S, F> Service for TracingService<S, F>
+where
+    S: Service,
+    F: Fn(&S::Request) -> Option<tracing::Span>,
+{
+    type Request = S::Request;
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = Either<S::Future, Instrumented<S::Future>>;
+
+    fn poll_ready(&mut self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(ctx)
+    }
+
+    fn call(&mut self, req: Self::Request) -> Self::Future {
+        let span = (self.make_span)(&req);
+        let _enter = span.as_ref().map(|s| s.enter());
+
+        let fut = self.inner.call(req);
+
+        // make a child span to track the future's execution
+        if let Some(span) = span
+            .clone()
+            .map(|span| tracing::span!(parent: &span, tracing::Level::INFO, "future"))
+        {
+            Either::Right(fut.instrument(span))
+        } else {
+            Either::Left(fut)
+        }
+    }
+}
+
+/// A `Transform` implementation that wraps services with a [`TracingService`].
+///
+/// [`TracingService`]: struct.TracingService.html
+pub struct TracingTransform<S, U, F> {
+    make_span: F,
+    _p: PhantomData<fn(S, U)>,
+}
+
+impl<S, U, F> TracingTransform<S, U, F> {
+    pub fn new(make_span: F) -> Self {
+        TracingTransform {
+            make_span,
+            _p: PhantomData,
+        }
+    }
+}
+
+impl<S, U, F> Transform<S> for TracingTransform<S, U, F>
+where
+    S: Service,
+    U: ServiceFactory<
+        Request = S::Request,
+        Response = S::Response,
+        Error = S::Error,
+        Service = S,
+    >,
+    F: Fn(&S::Request) -> Option<tracing::Span> + Clone,
+{
+    type Request = S::Request;
+    type Response = S::Response;
+    type Error = S::Error;
+    type Transform = TracingService<S, F>;
+    type InitError = U::InitError;
+    type Future = Ready<Result<Self::Transform, Self::InitError>>;
+
+    fn new_transform(&self, service: S) -> Self::Future {
+        ok(TracingService::new(service, self.make_span.clone()))
+    }
+}
+
+/// Wraps the provided service factory with a transform that automatically
+/// enters/exits the given span.
+///
+/// The span to be entered/exited can be provided via a closure. The closure
+/// is passed in a reference to the request being handled by the service.
+///
+/// For example:
+/// ```rust,ignore
+/// let traced_service = trace(
+///     web_service,
+///     |req: &Request| Some(span!(Level::INFO, "request", req.id))
+/// );
+/// ```
+pub fn trace<S, U, F>(
+    service_factory: U,
+    make_span: F,
+) -> ApplyTransform<TracingTransform<S::Service, S, F>, S>
+where
+    S: ServiceFactory,
+    F: Fn(&S::Request) -> Option<tracing::Span> + Clone,
+    U: IntoServiceFactory<S>,
+{
+    apply(
+        TracingTransform::new(make_span),
+        service_factory.into_factory(),
+    )
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    use std::cell::RefCell;
+    use std::collections::{BTreeMap, BTreeSet};
+    use std::sync::{Arc, RwLock};
+
+    use actix_service::{fn_factory, fn_service};
+    use slab::Slab;
+    use tracing::{span, Event, Level, Metadata, Subscriber};
+
+    thread_local! {
+        static SPAN: RefCell<Vec<span::Id>> = RefCell::new(Vec::new());
+    }
+
+    #[derive(Default)]
+    struct Stats {
+        entered_spans: BTreeSet<u64>,
+        exited_spans: BTreeSet<u64>,
+        events_count: BTreeMap<u64, usize>,
+    }
+
+    #[derive(Default)]
+    struct Inner {
+        spans: Slab<&'static Metadata<'static>>,
+        stats: Stats,
+    }
+
+    #[derive(Clone, Default)]
+    struct TestSubscriber {
+        inner: Arc<RwLock<Inner>>,
+    }
+
+    impl Subscriber for TestSubscriber {
+        fn enabled(&self, _metadata: &Metadata<'_>) -> bool {
+            true
+        }
+
+        fn new_span(&self, span: &span::Attributes<'_>) -> span::Id {
+            let id = self.inner.write().unwrap().spans.insert(span.metadata());
+            span::Id::from_u64(id as u64 + 1)
+        }
+
+        fn record(&self, _span: &span::Id, _values: &span::Record<'_>) {}
+
+        fn record_follows_from(&self, _span: &span::Id, _follows: &span::Id) {}
+
+        fn event(&self, event: &Event<'_>) {
+            let id = event
+                .parent()
+                .cloned()
+                .or_else(|| SPAN.with(|current_span| current_span.borrow().last().cloned()))
+                .unwrap();
+
+            *self
+                .inner
+                .write()
+                .unwrap()
+                .stats
+                .events_count
+                .entry(id.into_u64())
+                .or_insert(0) += 1;
+        }
+
+        fn enter(&self, span: &span::Id) {
+            self.inner
+                .write()
+                .unwrap()
+                .stats
+                .entered_spans
+                .insert(span.into_u64());
+
+            SPAN.with(|current_span| {
+                current_span.borrow_mut().push(span.clone());
+            });
+        }
+
+        fn exit(&self, span: &span::Id) {
+            self.inner
+                .write()
+                .unwrap()
+                .stats
+                .exited_spans
+                .insert(span.into_u64());
+
+            // we are guaranteed that on any given thread, spans are exited in reverse order
+            SPAN.with(|current_span| {
+                let leaving = current_span
+                    .borrow_mut()
+                    .pop()
+                    .expect("told to exit span when not in span");
+                assert_eq!(
+                    &leaving, span,
+                    "told to exit span that was not most recently entered"
+                );
+            });
+        }
+    }
+
+    #[actix_rt::test]
+    async fn service_call() {
+        let service_factory = fn_factory(|| {
+            ok::<_, ()>(fn_service(|req: &'static str| {
+                tracing::event!(Level::TRACE, "It's happening - {}!", req);
+                ok::<_, ()>(())
+            }))
+        });
+
+        let subscriber = TestSubscriber::default();
+        let _guard = tracing::subscriber::set_default(subscriber.clone());
+
+        let span_svc = span!(Level::TRACE, "span_svc");
+        let trace_service_factory = trace(service_factory, |_: &&str| Some(span_svc.clone()));
+        let mut service = trace_service_factory.new_service(()).await.unwrap();
+        service.call("boo").await.unwrap();
+
+        let id = span_svc.id().unwrap().into_u64();
+        assert!(subscriber
+            .inner
+            .read()
+            .unwrap()
+            .stats
+            .entered_spans
+            .contains(&id));
+        assert!(subscriber
+            .inner
+            .read()
+            .unwrap()
+            .stats
+            .exited_spans
+            .contains(&id));
+        assert_eq!(subscriber.inner.read().unwrap().stats.events_count[&id], 1);
+    }
+}

actix-utils/CHANGES.md

@@ -1,5 +1,15 @@
 # Changes
 
+## [1.0.6] - 2020-01-08
+
+* Add `Clone` impl for `condition::Waiter`
+
+## [1.0.5] - 2020-01-08
+
+* Add `Condition` type.
+
+* Add `Pool` of one-shot's.
+
 ## [1.0.4] - 2019-12-20
 
 * Add methods to check `LocalWaker` registration state.

actix-utils/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-utils"
-version = "1.0.4"
+version = "1.0.6"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix utils - various actix net related services"
 keywords = ["network", "framework", "async", "futures"]
@@ -16,11 +16,13 @@ name = "actix_utils"
 path = "src/lib.rs"
 
 [dependencies]
-actix-service = "1.0.0"
+actix-service = "1.0.1"
 actix-rt = "1.0.0"
 actix-codec = "0.2.0"
+bitflags = "1.2"
 bytes = "0.5.3"
 either = "1.5.3"
 futures = "0.3.1"
 pin-project = "0.4.6"
 log = "0.4"
+slab = "0.4"

actix-utils/src/condition.rs

@@ -0,0 +1,127 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use slab::Slab;
+
+use crate::cell::Cell;
+use crate::task::LocalWaker;
+
+/// Condition allows to notify multiple receivers at the same time
+pub struct Condition(Cell<Inner>);
+
+struct Inner {
+    data: Slab<Option<LocalWaker>>,
+}
+
+impl Default for Condition {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Condition {
+    pub fn new() -> Condition {
+        Condition(Cell::new(Inner { data: Slab::new() }))
+    }
+
+    /// Get condition waiter
+    pub fn wait(&mut self) -> Waiter {
+        let token = self.0.get_mut().data.insert(None);
+        Waiter {
+            token,
+            inner: self.0.clone(),
+        }
+    }
+
+    /// Notify all waiters
+    pub fn notify(&self) {
+        let inner = self.0.get_ref();
+        for item in inner.data.iter() {
+            if let Some(waker) = item.1 {
+                waker.wake();
+            }
+        }
+    }
+}
+
+impl Drop for Condition {
+    fn drop(&mut self) {
+        self.notify()
+    }
+}
+
+#[must_use = "Waiter do nothing unless polled"]
+pub struct Waiter {
+    token: usize,
+    inner: Cell<Inner>,
+}
+
+impl Clone for Waiter {
+    fn clone(&self) -> Self {
+        let token = unsafe { self.inner.get_mut_unsafe() }.data.insert(None);
+        Waiter {
+            token,
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+impl Future for Waiter {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.get_mut();
+
+        let inner = unsafe { this.inner.get_mut().data.get_unchecked_mut(this.token) };
+        if inner.is_none() {
+            let waker = LocalWaker::default();
+            waker.register(cx.waker());
+            *inner = Some(waker);
+            Poll::Pending
+        } else if inner.as_mut().unwrap().register(cx.waker()) {
+            Poll::Pending
+        } else {
+            Poll::Ready(())
+        }
+    }
+}
+
+impl Drop for Waiter {
+    fn drop(&mut self) {
+        self.inner.get_mut().data.remove(self.token);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::future::lazy;
+
+    #[actix_rt::test]
+    async fn test_condition() {
+        let mut cond = Condition::new();
+        let mut waiter = cond.wait();
+        assert_eq!(
+            lazy(|cx| Pin::new(&mut waiter).poll(cx)).await,
+            Poll::Pending
+        );
+        cond.notify();
+        waiter.await;
+
+        let mut waiter = cond.wait();
+        assert_eq!(
+            lazy(|cx| Pin::new(&mut waiter).poll(cx)).await,
+            Poll::Pending
+        );
+        let mut waiter2 = waiter.clone();
+        assert_eq!(
+            lazy(|cx| Pin::new(&mut waiter2).poll(cx)).await,
+            Poll::Pending
+        );
+
+        drop(cond);
+        waiter.await;
+        waiter2.await;
+    }
+}

actix-utils/src/lib.rs

@@ -3,6 +3,7 @@
 #![allow(clippy::type_complexity)]
 
 mod cell;
+pub mod condition;
 pub mod counter;
 pub mod either;
 pub mod framed;

actix-utils/src/oneshot.rs

@@ -4,6 +4,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 pub use futures::channel::oneshot::Canceled;
+use slab::Slab;
 
 use crate::cell::Cell;
 use crate::task::LocalWaker;
@@ -21,6 +22,11 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
     (tx, rx)
 }
 
+/// Creates a new futures-aware, pool of one-shot's.
+pub fn pool<T>() -> Pool<T> {
+    Pool(Cell::new(Slab::new()))
+}
+
 /// Represents the completion half of a oneshot through which the result of a
 /// computation is signaled.
 #[derive(Debug)]
@@ -77,9 +83,7 @@ impl<T> Sender<T> {
 
 impl<T> Drop for Sender<T> {
     fn drop(&mut self) {
-        if self.inner.strong_count() == 2 {
-            self.inner.get_ref().rx_task.wake();
-        };
+        self.inner.get_ref().rx_task.wake();
     }
 }
 
@@ -104,6 +108,148 @@ impl<T> Future for Receiver<T> {
     }
 }
 
+/// Futures-aware, pool of one-shot's.
+pub struct Pool<T>(Cell<Slab<PoolInner<T>>>);
+
+bitflags::bitflags! {
+    pub struct Flags: u8 {
+        const SENDER = 0b0000_0001;
+        const RECEIVER = 0b0000_0010;
+    }
+}
+
+#[derive(Debug)]
+struct PoolInner<T> {
+    flags: Flags,
+    value: Option<T>,
+    waker: LocalWaker,
+}
+
+impl<T> Pool<T> {
+    pub fn channel(&mut self) -> (PSender<T>, PReceiver<T>) {
+        let token = self.0.get_mut().insert(PoolInner {
+            flags: Flags::all(),
+            value: None,
+            waker: LocalWaker::default(),
+        });
+
+        (
+            PSender {
+                token,
+                inner: self.0.clone(),
+            },
+            PReceiver {
+                token,
+                inner: self.0.clone(),
+            },
+        )
+    }
+}
+
+impl<T> Clone for Pool<T> {
+    fn clone(&self) -> Self {
+        Pool(self.0.clone())
+    }
+}
+
+/// Represents the completion half of a oneshot through which the result of a
+/// computation is signaled.
+#[derive(Debug)]
+pub struct PSender<T> {
+    token: usize,
+    inner: Cell<Slab<PoolInner<T>>>,
+}
+
+/// A future representing the completion of a computation happening elsewhere in
+/// memory.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless polled"]
+pub struct PReceiver<T> {
+    token: usize,
+    inner: Cell<Slab<PoolInner<T>>>,
+}
+
+// The oneshots do not ever project Pin to the inner T
+impl<T> Unpin for PReceiver<T> {}
+impl<T> Unpin for PSender<T> {}
+
+impl<T> PSender<T> {
+    /// Completes this oneshot with a successful result.
+    ///
+    /// This function will consume `self` and indicate to the other end, the
+    /// `Receiver`, that the error provided is the result of the computation this
+    /// represents.
+    ///
+    /// If the value is successfully enqueued for the remote end to receive,
+    /// then `Ok(())` is returned. If the receiving end was dropped before
+    /// this function was called, however, then `Err` is returned with the value
+    /// provided.
+    pub fn send(mut self, val: T) -> Result<(), T> {
+        let inner = unsafe { self.inner.get_mut().get_unchecked_mut(self.token) };
+
+        if inner.flags.contains(Flags::RECEIVER) {
+            inner.value = Some(val);
+            inner.waker.wake();
+            Ok(())
+        } else {
+            Err(val)
+        }
+    }
+
+    /// Tests to see whether this `Sender`'s corresponding `Receiver`
+    /// has gone away.
+    pub fn is_canceled(&self) -> bool {
+        !unsafe { self.inner.get_ref().get_unchecked(self.token) }
+            .flags
+            .contains(Flags::RECEIVER)
+    }
+}
+
+impl<T> Drop for PSender<T> {
+    fn drop(&mut self) {
+        let inner = unsafe { self.inner.get_mut().get_unchecked_mut(self.token) };
+        if inner.flags.contains(Flags::RECEIVER) {
+            inner.waker.wake();
+            inner.flags.remove(Flags::SENDER);
+        } else {
+            self.inner.get_mut().remove(self.token);
+        }
+    }
+}
+
+impl<T> Drop for PReceiver<T> {
+    fn drop(&mut self) {
+        let inner = unsafe { self.inner.get_mut().get_unchecked_mut(self.token) };
+        if inner.flags.contains(Flags::SENDER) {
+            inner.flags.remove(Flags::RECEIVER);
+        } else {
+            self.inner.get_mut().remove(self.token);
+        }
+    }
+}
+
+impl<T> Future for PReceiver<T> {
+    type Output = Result<T, Canceled>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.get_mut();
+        let inner = unsafe { this.inner.get_mut().get_unchecked_mut(this.token) };
+
+        // If we've got a value, then skip the logic below as we're done.
+        if let Some(val) = inner.value.take() {
+            return Poll::Ready(Ok(val));
+        }
+
+        // Check if sender is dropped and return error if it is.
+        if !inner.flags.contains(Flags::SENDER) {
+            Poll::Ready(Err(Canceled))
+        } else {
+            inner.waker.register(cx.waker());
+            Poll::Pending
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -135,4 +281,31 @@ mod tests {
         drop(tx);
         assert!(rx.await.is_err());
     }
+
+    #[actix_rt::test]
+    async fn test_pool() {
+        let (tx, rx) = pool().channel();
+        tx.send("test").unwrap();
+        assert_eq!(rx.await.unwrap(), "test");
+
+        let (tx, rx) = pool().channel();
+        assert!(!tx.is_canceled());
+        drop(rx);
+        assert!(tx.is_canceled());
+        assert!(tx.send("test").is_err());
+
+        let (tx, rx) = pool::<&'static str>().channel();
+        drop(tx);
+        assert!(rx.await.is_err());
+
+        let (tx, mut rx) = pool::<&'static str>().channel();
+        assert_eq!(lazy(|cx| Pin::new(&mut rx).poll(cx)).await, Poll::Pending);
+        tx.send("test").unwrap();
+        assert_eq!(rx.await.unwrap(), "test");
+
+        let (tx, mut rx) = pool::<&'static str>().channel();
+        assert_eq!(lazy(|cx| Pin::new(&mut rx).poll(cx)).await, Poll::Pending);
+        drop(tx);
+        assert!(rx.await.is_err());
+    }
 }

actix-utils/src/order.rs

@@ -242,7 +242,7 @@ mod tests {
             let rx2 = rx2;
             let rx3 = rx3;
             let tx_stop = tx_stop;
-            let _ = actix_rt::System::new("test").block_on(async {
+            actix_rt::System::new("test").block_on(async {
                 let mut srv = InOrderService::new(Srv);
 
                 let _ = lazy(|cx| srv.poll_ready(cx)).await;
@@ -251,7 +251,7 @@ mod tests {
                 let res3 = srv.call(rx3);
 
                 actix_rt::spawn(async move {
-                    let _ = poll_fn(|cx| {
+                    poll_fn(|cx| {
                         let _ = srv.poll_ready(cx);
                         Poll::<()>::Pending
                     })

router/CHANGES.md

@@ -1,8 +1,8 @@
 # Changes
 
-## [0.3.0] - 2019-12-31
+## [0.2.4] - 2019-12-31
 
-* Support named parameters for `ResourceDef::resource_path()` in form of `((&k, &v), ...)`
+* Add `ResourceDef::resource_path_named()` path generation method
 
 ## [0.2.3] - 2019-12-25
 

router/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-router"
-version = "0.3.0"
+version = "0.2.4"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Path router"
 keywords = ["actix"]
@@ -18,11 +18,10 @@ path = "src/lib.rs"
 default = ["http"]
 
 [dependencies]
-bytestring = "0.1.2"
-either = "1.5.3"
 regex = "1.3.1"
-log = "0.4.8"
 serde = "1.0.104"
+bytestring = "0.1.2"
+log = "0.4.8"
 http = { version="0.2.0", optional=true }
 
 [dev-dependencies]

router/src/lib.rs

@@ -4,8 +4,6 @@ mod path;
 mod resource;
 mod router;
 
-use either::Either;
-
 pub use self::de::PathDeserializer;
 pub use self::path::Path;
 pub use self::resource::ResourceDef;
@@ -37,127 +35,6 @@ impl ResourcePath for bytestring::ByteString {
     }
 }
 
-pub trait ResourceElements {
-    fn elements<F, R>(self, for_each: F) -> Option<R>
-    where
-        F: FnMut(Either<&str, (&str, &str)>) -> Option<R>;
-}
-
-impl<'a, T: AsRef<str>> ResourceElements for &'a [T] {
-    fn elements<F, R>(self, mut for_each: F) -> Option<R>
-    where
-        F: FnMut(Either<&str, (&str, &str)>) -> Option<R>,
-    {
-        for t in self {
-            if let Some(res) = for_each(Either::Left(t.as_ref())) {
-                return Some(res);
-            }
-        }
-        None
-    }
-}
-
-impl<'a, U, I> ResourceElements for &'a U
-where
-    &'a U: IntoIterator<Item = I>,
-    I: AsRef<str>,
-{
-    fn elements<F, R>(self, mut for_each: F) -> Option<R>
-    where
-        F: FnMut(Either<&str, (&str, &str)>) -> Option<R>,
-    {
-        for t in self.into_iter() {
-            if let Some(res) = for_each(Either::Left(t.as_ref())) {
-                return Some(res);
-            }
-        }
-        None
-    }
-}
-
-impl<I> ResourceElements for Vec<I>
-where
-    I: AsRef<str>,
-{
-    fn elements<F, R>(self, mut for_each: F) -> Option<R>
-    where
-        F: FnMut(Either<&str, (&str, &str)>) -> Option<R>,
-    {
-        for t in self.iter() {
-            if let Some(res) = for_each(Either::Left(t.as_ref())) {
-                return Some(res);
-            }
-        }
-        None
-    }
-}
-
-impl<'a, K, V, S> ResourceElements for std::collections::HashMap<K, V, S>
-where
-    K: AsRef<str>,
-    V: AsRef<str>,
-    S: std::hash::BuildHasher,
-{
-    fn elements<F, R>(self, mut for_each: F) -> Option<R>
-    where
-        F: FnMut(Either<&str, (&str, &str)>) -> Option<R>,
-    {
-        for t in self.iter() {
-            if let Some(res) = for_each(Either::Right((t.0.as_ref(), t.1.as_ref()))) {
-                return Some(res);
-            }
-        }
-        None
-    }
-}
-
-#[rustfmt::skip]
-mod _m {
-use super::*;
-// macro_rules! elements_tuple ({ $(($n:tt, $T:ident)),+} => {
-//     impl<$($T: AsRef<str>,)+> ResourceElements for ($($T,)+) {
-//         fn elements<F_, R_>(self, mut for_each: F_) -> Option<R_>
-//         where
-//             F_: FnMut(Either<&str, (&str, &str)>) -> Option<R_>,
-//         {
-//             $(
-//                 if let Some(res) = for_each(Either::Left(self.$n.as_ref())) {
-//                     return Some(res)
-//                 }
-//             )+
-//                 None
-//         }
-//     }
-// });
-
-macro_rules! elements_2tuple ({ $(($n:tt, $V:ident)),+} => {
-    impl<'a, $($V: AsRef<str>,)+> ResourceElements for ($((&'a str, $V),)+) {
-        fn elements<F_, R_>(self, mut for_each: F_) -> Option<R_>
-        where
-            F_: FnMut(Either<&str, (&str, &str)>) -> Option<R_>,
-        {
-            $(
-                if let Some(res) = for_each(Either::Right((self.$n.0, self.$n.1.as_ref()))) {
-                    return Some(res)
-                }
-            )+
-                None
-        }
-    }
-});
-
-elements_2tuple!((0, A));
-elements_2tuple!((0, A), (1, B));
-elements_2tuple!((0, A), (1, B), (2, C));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D), (4, E));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D), (4, E), (5, F));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D), (4, E), (5, F), (6, G));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D), (4, E), (5, F), (6, G), (7, H));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D), (4, E), (5, F), (6, G), (7, H), (8, I));
-elements_2tuple!((0, A), (1, B), (2, C), (3, D), (4, E), (5, F), (6, G), (7, H), (8, I), (9, J));
-}
-
 /// Helper trait for type that could be converted to path pattern
 pub trait IntoPattern {
     /// Signle patter
@@ -192,7 +69,7 @@ impl<'a> IntoPattern for &'a str {
     }
 
     fn patterns(&self) -> Vec<String> {
-        vec![self.to_string()]
+        vec![(*self).to_string()]
     }
 }
 
@@ -202,7 +79,7 @@ impl<T: AsRef<str>> IntoPattern for Vec<T> {
     }
 
     fn patterns(&self) -> Vec<String> {
-        self.into_iter().map(|v| v.as_ref().to_string()).collect()
+        self.iter().map(|v| v.as_ref().to_string()).collect()
     }
 }
 

router/src/resource.rs

@@ -5,7 +5,7 @@ use std::hash::{Hash, Hasher};
 use regex::{escape, Regex, RegexSet};
 
 use crate::path::{Path, PathItem};
-use crate::{IntoPattern, Resource, ResourceElements, ResourcePath};
+use crate::{IntoPattern, Resource, ResourcePath};
 
 const MAX_DYNAMIC_SEGMENTS: usize = 16;
 
@@ -294,7 +294,7 @@ impl ResourceDef {
                     return false;
                 }
                 for idx in 0..idx {
-                    path.add(names[idx].clone(), segments[idx]);
+                    path.add(names[idx], segments[idx]);
                 }
                 path.skip((pos + len) as u16);
                 true
@@ -326,7 +326,7 @@ impl ResourceDef {
                         return false;
                     }
                     for idx in 0..idx {
-                        path.add(names[idx].clone(), segments[idx]);
+                        path.add(names[idx], segments[idx]);
                     }
                     path.skip((pos + len) as u16);
                     true
@@ -413,7 +413,7 @@ impl ResourceDef {
 
                 let path = res.resource_path();
                 for idx in 0..idx {
-                    path.add(names[idx].clone(), segments[idx]);
+                    path.add(names[idx], segments[idx]);
                 }
                 path.skip((pos + len) as u16);
                 true
@@ -452,7 +452,7 @@ impl ResourceDef {
 
                     let path = res.resource_path();
                     for idx in 0..idx {
-                        path.add(names[idx].clone(), segments[idx]);
+                        path.add(names[idx], segments[idx]);
                     }
                     path.skip((pos + len) as u16);
                     true
@@ -464,66 +464,61 @@ impl ResourceDef {
     }
 
     /// Build resource path from elements. Returns `true` on success.
-    pub fn resource_path<U: ResourceElements>(&self, path: &mut String, data: U) -> bool {
+    pub fn resource_path<U, I>(&self, path: &mut String, elements: &mut U) -> bool
+    where
+        U: Iterator<Item = I>,
+        I: AsRef<str>,
+    {
         match self.tp {
             PatternType::Prefix(ref p) => path.push_str(p),
             PatternType::Static(ref p) => path.push_str(p),
             PatternType::Dynamic(..) => {
-                let mut iter = self.elements.iter();
-                let mut map = HashMap::new();
-
-                let result = data.elements(|item| match item {
-                    either::Either::Left(val) => loop {
-                        if let Some(el) = iter.next() {
-                            match *el {
-                                PatternElement::Str(ref s) => path.push_str(s),
-                                PatternElement::Var(_) => {
-                                    path.push_str(val.as_ref());
-                                    return None;
-                                }
-                            }
-                        } else {
-                            return Some(false);
-                        }
-                    },
-                    either::Either::Right((name, val)) => {
-                        map.insert(name.to_string(), val.to_string());
-                        None
-                    }
-                });
-
-                if result.is_some() {
-                    return true;
-                } else {
-                    if map.is_empty() {
-                        // push static sections
-                        loop {
-                            if let Some(el) = iter.next() {
-                                match *el {
-                                    PatternElement::Str(ref s) => path.push_str(s),
-                                    PatternElement::Var(_) => {
-                                        return false;
-                                    }
-                                }
+                for el in &self.elements {
+                    match *el {
+                        PatternElement::Str(ref s) => path.push_str(s),
+                        PatternElement::Var(_) => {
+                            if let Some(val) = elements.next() {
+                                path.push_str(val.as_ref())
                             } else {
-                                break;
-                            }
-                        }
-                    } else {
-                        for el in iter {
-                            match *el {
-                                PatternElement::Str(ref s) => path.push_str(s),
-                                PatternElement::Var(ref name) => {
-                                    if let Some(val) = map.get(name) {
-                                        path.push_str(val);
-                                        continue;
-                                    }
-                                    return false;
-                                }
+                                return false;
+                            }
+                        }
+                    }
+                }
+            }
+            PatternType::DynamicSet(..) => {
+                return false;
+            }
+        }
+        true
+    }
+
+    /// Build resource path from elements. Returns `true` on success.
+    pub fn resource_path_named<K, V, S>(
+        &self,
+        path: &mut String,
+        elements: &HashMap<K, V, S>,
+    ) -> bool
+    where
+        K: std::borrow::Borrow<str> + Eq + Hash,
+        V: AsRef<str>,
+        S: std::hash::BuildHasher,
+    {
+        match self.tp {
+            PatternType::Prefix(ref p) => path.push_str(p),
+            PatternType::Static(ref p) => path.push_str(p),
+            PatternType::Dynamic(..) => {
+                for el in &self.elements {
+                    match *el {
+                        PatternElement::Str(ref s) => path.push_str(s),
+                        PatternElement::Var(ref name) => {
+                            if let Some(val) = elements.get(name) {
+                                path.push_str(val.as_ref())
+                            } else {
+                                return false;
                             }
                         }
                     }
-                    return true;
                 }
             }
             PatternType::DynamicSet(..) => {
@@ -673,7 +668,6 @@ pub(crate) fn insert_slash(path: &str) -> String {
 mod tests {
     use super::*;
     use http::Uri;
-    use std::collections::HashMap;
     use std::convert::TryFrom;
 
     #[test]
@@ -740,6 +734,7 @@ mod tests {
         assert_eq!(path.get("id").unwrap(), "012345");
     }
 
+    #[allow(clippy::cognitive_complexity)]
     #[test]
     fn test_dynamic_set() {
         let re = ResourceDef::new(vec![
@@ -905,66 +900,45 @@ mod tests {
     fn test_resource_path() {
         let mut s = String::new();
         let resource = ResourceDef::new("/user/{item1}/test");
-        assert!(resource.resource_path(&mut s, &["user1"]));
-        assert_eq!(s, "/user/user1/test");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, (("item1", "user1"),)));
+        assert!(resource.resource_path(&mut s, &mut (&["user1"]).iter()));
         assert_eq!(s, "/user/user1/test");
 
         let mut s = String::new();
         let resource = ResourceDef::new("/user/{item1}/{item2}/test");
-        assert!(resource.resource_path(&mut s, &["item", "item2"]));
-        assert_eq!(s, "/user/item/item2/test");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, (("item1", "item"), ("item2", "item2"))));
+        assert!(resource.resource_path(&mut s, &mut (&["item", "item2"]).iter()));
         assert_eq!(s, "/user/item/item2/test");
 
         let mut s = String::new();
         let resource = ResourceDef::new("/user/{item1}/{item2}");
-        assert!(resource.resource_path(&mut s, &["item", "item2"]));
-        assert_eq!(s, "/user/item/item2");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, (("item1", "item"), ("item2", "item2"))));
+        assert!(resource.resource_path(&mut s, &mut (&["item", "item2"]).iter()));
         assert_eq!(s, "/user/item/item2");
 
         let mut s = String::new();
         let resource = ResourceDef::new("/user/{item1}/{item2}/");
-        assert!(resource.resource_path(&mut s, &["item", "item2"]));
+        assert!(resource.resource_path(&mut s, &mut (&["item", "item2"]).iter()));
         assert_eq!(s, "/user/item/item2/");
 
         let mut s = String::new();
-        assert!(!resource.resource_path(&mut s, &["item"]));
+        assert!(!resource.resource_path(&mut s, &mut (&["item"]).iter()));
 
         let mut s = String::new();
-        assert!(resource.resource_path(&mut s, &["item", "item2"]));
+        assert!(resource.resource_path(&mut s, &mut (&["item", "item2"]).iter()));
         assert_eq!(s, "/user/item/item2/");
-        assert!(!resource.resource_path(&mut s, &["item"]));
+        assert!(!resource.resource_path(&mut s, &mut (&["item"]).iter()));
 
         let mut s = String::new();
-        assert!(resource.resource_path(&mut s, vec!["item", "item2"]));
-        assert_eq!(s, "/user/item/item2/");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, &vec!["item", "item2"]));
-        assert_eq!(s, "/user/item/item2/");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, &vec!["item", "item2"][..]));
+        assert!(resource.resource_path(&mut s, &mut vec!["item", "item2"].into_iter()));
         assert_eq!(s, "/user/item/item2/");
 
         let mut map = HashMap::new();
         map.insert("item1", "item");
+
+        let mut s = String::new();
+        assert!(!resource.resource_path_named(&mut s, &map));
+
+        let mut s = String::new();
         map.insert("item2", "item2");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, map));
-        assert_eq!(s, "/user/item/item2/");
-
-        let mut s = String::new();
-        assert!(resource.resource_path(&mut s, (("item1", "item"), ("item2", "item2"))));
+        assert!(resource.resource_path_named(&mut s, &map));
         assert_eq!(s, "/user/item/item2/");
     }
 }

router/src/router.rs

@@ -104,6 +104,7 @@ mod tests {
     use crate::path::Path;
     use crate::router::{ResourceId, Router};
 
+    #[allow(clippy::cognitive_complexity)]
     #[test]
     fn test_recognizer_1() {
         let mut router = Router::<usize>::build();

string/CHANGES.md

@@ -1,5 +1,13 @@
 # Changes
 
+## [0.1.4] - 2020-01-14
+
+* Fix `AsRef<str>` impl
+
+## [0.1.3] - 2020-01-13
+
+* Add `PartialEq<T: AsRef<str>>`, `AsRef<[u8]>` impls
+
 ## [0.1.2] - 2019-12-22
 
 * Fix `new()` method

string/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bytestring"
-version = "0.1.2"
+version = "0.1.4"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "A UTF-8 encoded string with Bytes as a storage"
 keywords = ["actix"]

string/src/lib.rs

@@ -7,7 +7,7 @@ use bytes::Bytes;
 /// A utf-8 encoded string with [`Bytes`] as a storage.
 ///
 /// [`Bytes`]: https://docs.rs/bytes/0.5.3/bytes/struct.Bytes.html
-#[derive(Clone, Eq, PartialEq, Ord, PartialOrd, Default)]
+#[derive(Clone, Eq, Ord, PartialOrd, Default)]
 pub struct ByteString(Bytes);
 
 impl ByteString {
@@ -43,6 +43,24 @@ impl PartialEq<str> for ByteString {
     }
 }
 
+impl<T: AsRef<str>> PartialEq<T> for ByteString {
+    fn eq(&self, other: &T) -> bool {
+        &self[..] == other.as_ref()
+    }
+}
+
+impl AsRef<[u8]> for ByteString {
+    fn as_ref(&self) -> &[u8] {
+        self.0.as_ref()
+    }
+}
+
+impl AsRef<str> for ByteString {
+    fn as_ref(&self) -> &str {
+        &*self
+    }
+}
+
 impl hash::Hash for ByteString {
     fn hash<H: hash::Hasher>(&self, state: &mut H) {
         (**self).hash(state);
@@ -147,6 +165,14 @@ mod test {
     use std::collections::hash_map::DefaultHasher;
     use std::hash::{Hash, Hasher};
 
+    #[test]
+    fn test_partial_eq() {
+        let s: ByteString = ByteString::from_static("test");
+        assert_eq!(s, "test");
+        assert_eq!(s, *"test");
+        assert_eq!(s, "test".to_string());
+    }
+
     #[test]
     fn test_new() {
         let _: ByteString = ByteString::new();
@@ -167,6 +193,8 @@ mod test {
     fn test_from_string() {
         let s: ByteString = "hello".to_string().into();
         assert_eq!(&s, "hello");
+        let t: &str = s.as_ref();
+        assert_eq!(t, "hello");
     }
 
     #[test]
@@ -176,7 +204,7 @@ mod test {
 
     #[test]
     fn test_from_static_str() {
-        const _S: ByteString = ByteString::from_static("hello");
+        static _S: ByteString = ByteString::from_static("hello");
         let _ = ByteString::from_static("str");
     }