Merge pull request #142 from JohnTitor/new-server

server: Bump up to 1.0.3

.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,76 @@
+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' || github.event_name == 'pull_request')
+        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' || github.event_name == 'pull_request')
+        uses: codecov/codecov-action@v1
+        with:
+          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,39 @@
+name: CI (Windows-mingw)
+
+on: [push, pull_request]
+
+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 packages
+        run: |
+          msys2do pacman -Sy --noconfirm pacman
+          msys2do pacman --noconfirm -S base-devel pkg-config
+
+      - name: check build
+        uses: actions-rs/cargo@v1
+        with:
+          command: check
+          args: --all --bins --examples --tests

.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

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-codec/Cargo.toml

@@ -19,8 +19,9 @@ path = "src/lib.rs"
 [dependencies]
 bitflags = "1.2.1"
 bytes = "0.5.2"
-futures-core = "0.3.1"
-futures-sink = "0.3.1"
+futures-core = { version = "0.3.4", default-features = false }
+futures-sink = { version = "0.3.4", default-features = false }
 tokio = { version = "0.2.4", default-features=false }
 tokio-util = { version = "0.2.0", default-features=false, features=["codec"] }
-log = "0.4"
+log = "0.4"
+pin-project = "0.4.8"

actix-codec/src/framed.rs

@@ -5,6 +5,7 @@ use std::{fmt, io};
 use bytes::{Buf, BytesMut};
 use futures_core::{ready, Stream};
 use futures_sink::Sink;
+use pin_project::pin_project;
 
 use crate::{AsyncRead, AsyncWrite, Decoder, Encoder};
 
@@ -20,7 +21,9 @@ bitflags::bitflags! {
 
 /// A unified `Stream` and `Sink` interface to an underlying I/O object, using
 /// the `Encoder` and `Decoder` traits to encode and decode frames.
+#[pin_project]
 pub struct Framed<T, U> {
+    #[pin]
     io: T,
     codec: U,
     flags: Flags,
@@ -28,8 +31,6 @@ pub struct Framed<T, U> {
     write_buf: BytesMut,
 }
 
-impl<T, U> Unpin for Framed<T, U> {}
-
 impl<T, U> Framed<T, U>
 where
     T: AsyncRead + AsyncWrite,
@@ -185,17 +186,18 @@ impl<T, U> Framed<T, U> {
 
 impl<T, U> Framed<T, U> {
     /// Serialize item and Write to the inner buffer
-    pub fn write(&mut self, item: <U as Encoder>::Item) -> Result<(), <U as Encoder>::Error>
+    pub fn write(mut self: Pin<&mut Self>, item: <U as Encoder>::Item) -> Result<(), <U as Encoder>::Error>
     where
         T: AsyncWrite,
         U: Encoder,
     {
-        let remaining = self.write_buf.capacity() - self.write_buf.len();
+        let this = self.as_mut().project();
+        let remaining = this.write_buf.capacity() - this.write_buf.len();
         if remaining < LW {
-            self.write_buf.reserve(HW - remaining);
+            this.write_buf.reserve(HW - remaining);
         }
 
-        self.codec.encode(item, &mut self.write_buf)?;
+        this.codec.encode(item, this.write_buf)?;
         Ok(())
     }
 
@@ -207,21 +209,22 @@ impl<T, U> Framed<T, U> {
     }
 
     /// Try to read underlying I/O stream and decode item.
-    pub fn next_item(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<U::Item, U::Error>>>
+    pub fn next_item(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Result<U::Item, U::Error>>>
     where
         T: AsyncRead,
         U: Decoder,
     {
         loop {
+            let mut this = self.as_mut().project();
             // Repeatedly call `decode` or `decode_eof` as long as it is
             // "readable". Readable is defined as not having returned `None`. If
             // the upstream has returned EOF, and the decoder is no longer
             // readable, it can be assumed that the decoder will never become
             // readable again, at which point the stream is terminated.
 
-            if self.flags.contains(Flags::READABLE) {
-                if self.flags.contains(Flags::EOF) {
-                    match self.codec.decode_eof(&mut self.read_buf) {
+            if this.flags.contains(Flags::READABLE) {
+                if this.flags.contains(Flags::EOF) {
+                    match this.codec.decode_eof(&mut this.read_buf) {
                         Ok(Some(frame)) => return Poll::Ready(Some(Ok(frame))),
                         Ok(None) => return Poll::Ready(None),
                         Err(e) => return Poll::Ready(Some(Err(e))),
@@ -230,7 +233,7 @@ impl<T, U> Framed<T, U> {
 
                 log::trace!("attempting to decode a frame");
 
-                match self.codec.decode(&mut self.read_buf) {
+                match this.codec.decode(&mut this.read_buf) {
                     Ok(Some(frame)) => {
                         log::trace!("frame decoded from buffer");
                         return Poll::Ready(Some(Ok(frame)));
@@ -239,45 +242,44 @@ impl<T, U> Framed<T, U> {
                     _ => (), // Need more data
                 }
 
-                self.flags.remove(Flags::READABLE);
+                this.flags.remove(Flags::READABLE);
             }
 
-            debug_assert!(!self.flags.contains(Flags::EOF));
+            debug_assert!(!this.flags.contains(Flags::EOF));
 
             // Otherwise, try to read more data and try again. Make sure we've got room
-            let remaining = self.read_buf.capacity() - self.read_buf.len();
+            let remaining = this.read_buf.capacity() - this.read_buf.len();
             if remaining < LW {
-                self.read_buf.reserve(HW - remaining)
+                this.read_buf.reserve(HW - remaining)
             }
-            let cnt = match unsafe {
-                Pin::new_unchecked(&mut self.io).poll_read_buf(cx, &mut self.read_buf)
-            } {
+            let cnt = match this.io.poll_read_buf(cx, &mut this.read_buf) {
                 Poll::Pending => return Poll::Pending,
                 Poll::Ready(Err(e)) => return Poll::Ready(Some(Err(e.into()))),
                 Poll::Ready(Ok(cnt)) => cnt,
             };
 
             if cnt == 0 {
-                self.flags.insert(Flags::EOF);
+                this.flags.insert(Flags::EOF);
             }
-            self.flags.insert(Flags::READABLE);
+            this.flags.insert(Flags::READABLE);
         }
     }
 
     /// Flush write buffer to underlying I/O stream.
-    pub fn flush(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
+    pub fn flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
     where
         T: AsyncWrite,
         U: Encoder,
     {
+        let mut this = self.as_mut().project();
         log::trace!("flushing framed transport");
 
-        while !self.write_buf.is_empty() {
-            log::trace!("writing; remaining={}", self.write_buf.len());
+        while !this.write_buf.is_empty() {
+            log::trace!("writing; remaining={}", this.write_buf.len());
 
-            let n = ready!(unsafe {
-                Pin::new_unchecked(&mut self.io).poll_write(cx, &self.write_buf)
-            })?;
+            let n = ready!(
+                this.io.as_mut().poll_write(cx, this.write_buf)
+            )?;
 
             if n == 0 {
                 return Poll::Ready(Err(io::Error::new(
@@ -288,26 +290,25 @@ impl<T, U> Framed<T, U> {
             }
 
             // remove written data
-            self.write_buf.advance(n);
+            this.write_buf.advance(n);
         }
 
         // Try flushing the underlying IO
-        ready!(unsafe { Pin::new_unchecked(&mut self.io).poll_flush(cx) })?;
+        ready!(this.io.poll_flush(cx))?;
 
         log::trace!("framed transport flushed");
         Poll::Ready(Ok(()))
     }
 
     /// Flush write buffer and shutdown underlying I/O stream.
-    pub fn close(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
+    pub fn close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), U::Error>>
     where
         T: AsyncWrite,
         U: Encoder,
     {
-        unsafe {
-            ready!(Pin::new_unchecked(&mut self.io).poll_flush(cx))?;
-            ready!(Pin::new_unchecked(&mut self.io).poll_shutdown(cx))?;
-        }
+        let mut this = self.as_mut().project();
+        ready!(this.io.as_mut().poll_flush(cx))?;
+        ready!(this.io.as_mut().poll_shutdown(cx))?;
         Poll::Ready(Ok(()))
     }
 }
@@ -319,7 +320,7 @@ where
 {
     type Item = Result<U::Item, U::Error>;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         self.next_item(cx)
     }
 }
@@ -341,21 +342,21 @@ where
     }
 
     fn start_send(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         item: <U as Encoder>::Item,
     ) -> Result<(), Self::Error> {
         self.write(item)
     }
 
     fn poll_flush(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), Self::Error>> {
         self.flush(cx)
     }
 
     fn poll_close(
-        mut self: Pin<&mut Self>,
+        self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), Self::Error>> {
         self.close(cx)

actix-connect/CHANGES.md

@@ -1,5 +1,29 @@
 # Changes
 
+## [2.0.0-alpha.3] - 2020-05-08
+
+### Fixed
+
+* Corrected spelling of `ConnectError::Unresolverd` to `ConnectError::Unresolved`
+
+## [2.0.0-alpha.2] - 2020-03-08
+
+### Changed
+
+* Update `trust-dns-proto` dependency to 0.19. [#116]
+* Update `trust-dns-resolver` dependency to 0.19. [#116]
+* `Address` trait is now required to have static lifetime. [#116]
+* `start_resolver` and `start_default_resolver` are now `async` and may return a `ConnectError`. [#116]
+
+[#116]: https://github.com/actix/actix-net/pull/116
+
+## [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

actix-connect/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-connect"
-version = "1.0.2"
+version = "2.0.0-alpha.3"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix connect - tcp connector service"
 keywords = ["network", "framework", "async", "futures"]
@@ -37,19 +37,19 @@ actix-utils = "1.0.6"
 actix-rt = "1.0.0"
 derive_more = "0.99.2"
 either = "1.5.3"
-futures = "0.3.1"
+futures-util = { version = "0.3.4", default-features = false }
 http = { version = "0.2.0", optional = true }
 log = "0.4"
-trust-dns-proto = "=0.18.0-alpha.2"
-trust-dns-resolver = "=0.18.0-alpha.2"
+trust-dns-proto = { version = "0.19", default-features = false, features = ["tokio-runtime"] }
+trust-dns-resolver = { version = "0.19", default-features = false, features = ["tokio-runtime", "system-config"] }
 
 # openssl
 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/connect.rs

@@ -6,7 +6,7 @@ use std::net::SocketAddr;
 use either::Either;
 
 /// Connect request
-pub trait Address: Unpin {
+pub trait Address: Unpin + 'static {
     /// Host name of the request
     fn host(&self) -> &str;
 

actix-connect/src/connector.rs

@@ -8,7 +8,7 @@ use std::task::{Context, Poll};
 
 use actix_rt::net::TcpStream;
 use actix_service::{Service, ServiceFactory};
-use futures::future::{err, ok, BoxFuture, Either, FutureExt, Ready};
+use futures_util::future::{err, ok, BoxFuture, Either, FutureExt, Ready};
 
 use super::connect::{Address, Connect, Connection};
 use super::error::ConnectError;
@@ -88,7 +88,7 @@ impl<T: Address> Service for TcpConnector<T> {
             Either::Left(TcpConnectorResponse::new(req, port, addr))
         } else {
             error!("TCP connector: got unresolved address");
-            Either::Right(err(ConnectError::Unresolverd))
+            Either::Right(err(ConnectError::Unresolved))
         }
     }
 }

actix-connect/src/error.rs

@@ -18,7 +18,7 @@ pub enum ConnectError {
 
     /// Unresolved host name
     #[display(fmt = "Connector received `Connect` method with unresolved host")]
-    Unresolverd,
+    Unresolved,
 
     /// Connection io error
     #[display(fmt = "{}", _0)]

actix-connect/src/lib.rs

@@ -25,7 +25,7 @@ use actix_rt::{net::TcpStream, Arbiter};
 use actix_service::{pipeline, pipeline_factory, Service, ServiceFactory};
 use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
 use trust_dns_resolver::system_conf::read_system_conf;
-use trust_dns_resolver::AsyncResolver;
+use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
 
 pub mod resolver {
     pub use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
@@ -39,17 +39,18 @@ pub use self::error::ConnectError;
 pub use self::resolve::{Resolver, ResolverFactory};
 pub use self::service::{ConnectService, ConnectServiceFactory, TcpConnectService};
 
-pub fn start_resolver(cfg: ResolverConfig, opts: ResolverOpts) -> AsyncResolver {
-    let (resolver, bg) = AsyncResolver::new(cfg, opts);
-    actix_rt::spawn(bg);
-    resolver
+pub async fn start_resolver(
+    cfg: ResolverConfig,
+    opts: ResolverOpts,
+) -> Result<AsyncResolver, ConnectError> {
+    Ok(AsyncResolver::tokio(cfg, opts).await?)
 }
 
 struct DefaultResolver(AsyncResolver);
 
-pub(crate) fn get_default_resolver() -> AsyncResolver {
+pub(crate) async fn get_default_resolver() -> Result<AsyncResolver, ConnectError> {
     if Arbiter::contains_item::<DefaultResolver>() {
-        Arbiter::get_item(|item: &DefaultResolver| item.0.clone())
+        Ok(Arbiter::get_item(|item: &DefaultResolver| item.0.clone()))
     } else {
         let (cfg, opts) = match read_system_conf() {
             Ok((cfg, opts)) => (cfg, opts),
@@ -59,16 +60,15 @@ pub(crate) fn get_default_resolver() -> AsyncResolver {
             }
         };
 
-        let (resolver, bg) = AsyncResolver::new(cfg, opts);
-        actix_rt::spawn(bg);
+        let resolver = AsyncResolver::tokio(cfg, opts).await?;
 
         Arbiter::set_item(DefaultResolver(resolver.clone()));
-        resolver
+        Ok(resolver)
     }
 }
 
-pub fn start_default_resolver() -> AsyncResolver {
-    get_default_resolver()
+pub async fn start_default_resolver() -> Result<AsyncResolver, ConnectError> {
+    get_default_resolver().await
 }
 
 /// Create tcp connector service

actix-connect/src/resolve.rs

@@ -5,9 +5,9 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use actix_service::{Service, ServiceFactory};
-use futures::future::{ok, Either, Ready};
-use trust_dns_resolver::lookup_ip::LookupIpFuture;
-use trust_dns_resolver::{AsyncResolver, Background};
+use futures_util::future::{ok, Either, Ready};
+use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
+use trust_dns_resolver::{error::ResolveError, lookup_ip::LookupIp};
 
 use crate::connect::{Address, Connect};
 use crate::error::ConnectError;
@@ -106,7 +106,10 @@ impl<T: Address> Service for Resolver<T> {
     type Request = Connect<T>;
     type Response = Connect<T>;
     type Error = ConnectError;
-    type Future = Either<ResolverFuture<T>, Ready<Result<Connect<T>, Self::Error>>>;
+    type Future = Either<
+        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>,
+        Ready<Result<Connect<T>, Self::Error>>,
+    >;
 
     fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         Poll::Ready(Ok(()))
@@ -119,32 +122,48 @@ impl<T: Address> Service for Resolver<T> {
             req.addr = Some(either::Either::Left(SocketAddr::new(ip, req.port())));
             Either::Right(ok(req))
         } else {
-            trace!("DNS resolver: resolving host {:?}", req.host());
-            if self.resolver.is_none() {
-                self.resolver = Some(get_default_resolver());
-            }
-            Either::Left(ResolverFuture::new(req, self.resolver.as_ref().unwrap()))
+            let resolver = self.resolver.as_ref().map(AsyncResolver::clone);
+            Either::Left(Box::pin(async move {
+                trace!("DNS resolver: resolving host {:?}", req.host());
+                let resolver = if let Some(resolver) = resolver {
+                    resolver
+                } else {
+                    get_default_resolver()
+                        .await
+                        .expect("Failed to get default resolver")
+                };
+                ResolverFuture::new(req, &resolver).await
+            }))
         }
     }
 }
 
+type LookupIpFuture = Pin<Box<dyn Future<Output = Result<LookupIp, ResolveError>>>>;
+
 #[doc(hidden)]
 /// Resolver future
 pub struct ResolverFuture<T: Address> {
     req: Option<Connect<T>>,
-    lookup: Background<LookupIpFuture>,
+    lookup: LookupIpFuture,
 }
 
 impl<T: Address> ResolverFuture<T> {
     pub fn new(req: Connect<T>, resolver: &AsyncResolver) -> Self {
-        let lookup = if let Some(host) = req.host().splitn(2, ':').next() {
-            resolver.lookup_ip(host)
+        let host = if let Some(host) = req.host().splitn(2, ':').next() {
+            host
         } else {
-            resolver.lookup_ip(req.host())
+            req.host()
         };
 
+        // Clone data to be moved to the lookup future
+        let host_clone = host.to_owned();
+        let resolver_clone = resolver.clone();
+
         ResolverFuture {
-            lookup,
+            lookup: Box::pin(async move {
+                let resolver = resolver_clone;
+                resolver.lookup_ip(host_clone).await
+            }),
             req: Some(req),
         }
     }

actix-connect/src/service.rs

@@ -5,8 +5,8 @@ use std::task::{Context, Poll};
 use actix_rt::net::TcpStream;
 use actix_service::{Service, ServiceFactory};
 use either::Either;
-use futures::future::{ok, Ready};
-use trust_dns_resolver::AsyncResolver;
+use futures_util::future::{ok, Ready};
+use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
 
 use crate::connect::{Address, Connect, Connection};
 use crate::connector::{TcpConnector, TcpConnectorFactory};

actix-connect/src/ssl/openssl.rs

@@ -10,8 +10,8 @@ pub use tokio_openssl::{HandshakeError, SslStream};
 use actix_codec::{AsyncRead, AsyncWrite};
 use actix_rt::net::TcpStream;
 use actix_service::{Service, ServiceFactory};
-use futures::future::{err, ok, Either, FutureExt, LocalBoxFuture, Ready};
-use trust_dns_resolver::AsyncResolver;
+use futures_util::future::{err, ok, Either, FutureExt, LocalBoxFuture, Ready};
+use trust_dns_resolver::TokioAsyncResolver as AsyncResolver;
 
 use crate::{
     Address, Connect, ConnectError, ConnectService, ConnectServiceFactory, Connection,
@@ -243,7 +243,7 @@ impl<T: Address> Future for OpensslConnectServiceResponse<T> {
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         if let Some(ref mut fut) = self.fut1 {
-            match futures::ready!(Pin::new(fut).poll(cx)) {
+            match futures_util::ready!(Pin::new(fut).poll(cx)) {
                 Ok(res) => {
                     let _ = self.fut1.take();
                     self.fut2 = Some(self.openssl.call(res));
@@ -253,7 +253,7 @@ impl<T: Address> Future for OpensslConnectServiceResponse<T> {
         }
 
         if let Some(ref mut fut) = self.fut2 {
-            match futures::ready!(Pin::new(fut).poll(cx)) {
+            match futures_util::ready!(Pin::new(fut).poll(cx)) {
                 Ok(connect) => Poll::Ready(Ok(connect.into_parts().0)),
                 Err(e) => Poll::Ready(Err(ConnectError::Io(io::Error::new(
                     io::ErrorKind::Other,

actix-connect/src/ssl/rustls.rs

@@ -10,7 +10,7 @@ pub use tokio_rustls::{client::TlsStream, rustls::ClientConfig};
 
 use actix_codec::{AsyncRead, AsyncWrite};
 use actix_service::{Service, ServiceFactory};
-use futures::future::{ok, Ready};
+use futures_util::future::{ok, Ready};
 use tokio_rustls::{Connect, TlsConnector};
 use webpki::DNSNameRef;
 
@@ -38,7 +38,7 @@ where
 {
     pub fn service(connector: Arc<ClientConfig>) -> RustlsConnectorService<T, U> {
         RustlsConnectorService {
-            connector: connector,
+            connector,
             _t: PhantomData,
         }
     }
@@ -126,7 +126,7 @@ where
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.get_mut();
         Poll::Ready(
-            futures::ready!(Pin::new(&mut this.fut).poll(cx)).map(|stream| {
+            futures_util::ready!(Pin::new(&mut this.fut).poll(cx)).map(|stream| {
                 let s = this.stream.take().unwrap();
                 trace!("SSL Handshake success: {:?}", s.host());
                 s.replace(stream).1

actix-connect/tests/test_connect.rs

@@ -5,7 +5,7 @@ use actix_rt::net::TcpStream;
 use actix_service::{fn_service, Service, ServiceFactory};
 use actix_testing::TestServer;
 use bytes::Bytes;
-use futures::SinkExt;
+use futures_util::sink::SinkExt;
 
 use actix_connect::resolver::{ResolverConfig, ResolverOpts};
 use actix_connect::Connect;
@@ -14,12 +14,10 @@ use actix_connect::Connect;
 #[actix_rt::test]
 async fn test_string() {
     let srv = TestServer::with(|| {
-        fn_service(|io: TcpStream| {
-            async {
-                let mut framed = Framed::new(io, BytesCodec);
-                framed.send(Bytes::from_static(b"test")).await?;
-                Ok::<_, io::Error>(())
-            }
+        fn_service(|io: TcpStream| async {
+            let mut framed = Framed::new(io, BytesCodec);
+            framed.send(Bytes::from_static(b"test")).await?;
+            Ok::<_, io::Error>(())
         })
     });
 
@@ -33,12 +31,10 @@ async fn test_string() {
 #[actix_rt::test]
 async fn test_rustls_string() {
     let srv = TestServer::with(|| {
-        fn_service(|io: TcpStream| {
-            async {
-                let mut framed = Framed::new(io, BytesCodec);
-                framed.send(Bytes::from_static(b"test")).await?;
-                Ok::<_, io::Error>(())
-            }
+        fn_service(|io: TcpStream| async {
+            let mut framed = Framed::new(io, BytesCodec);
+            framed.send(Bytes::from_static(b"test")).await?;
+            Ok::<_, io::Error>(())
         })
     });
 
@@ -51,16 +47,14 @@ async fn test_rustls_string() {
 #[actix_rt::test]
 async fn test_static_str() {
     let srv = TestServer::with(|| {
-        fn_service(|io: TcpStream| {
-            async {
-                let mut framed = Framed::new(io, BytesCodec);
-                framed.send(Bytes::from_static(b"test")).await?;
-                Ok::<_, io::Error>(())
-            }
+        fn_service(|io: TcpStream| async {
+            let mut framed = Framed::new(io, BytesCodec);
+            framed.send(Bytes::from_static(b"test")).await?;
+            Ok::<_, io::Error>(())
         })
     });
 
-    let resolver = actix_connect::start_default_resolver();
+    let resolver = actix_connect::start_default_resolver().await.unwrap();
     let mut conn = actix_connect::new_connector(resolver.clone());
 
     let con = conn.call(Connect::with("10", srv.addr())).await.unwrap();
@@ -75,17 +69,17 @@ async fn test_static_str() {
 #[actix_rt::test]
 async fn test_new_service() {
     let srv = TestServer::with(|| {
-        fn_service(|io: TcpStream| {
-            async {
-                let mut framed = Framed::new(io, BytesCodec);
-                framed.send(Bytes::from_static(b"test")).await?;
-                Ok::<_, io::Error>(())
-            }
+        fn_service(|io: TcpStream| async {
+            let mut framed = Framed::new(io, BytesCodec);
+            framed.send(Bytes::from_static(b"test")).await?;
+            Ok::<_, io::Error>(())
         })
     });
 
     let resolver =
-        actix_connect::start_resolver(ResolverConfig::default(), ResolverOpts::default());
+        actix_connect::start_resolver(ResolverConfig::default(), ResolverOpts::default())
+            .await
+            .unwrap();
 
     let factory = actix_connect::new_connector_factory(resolver);
 
@@ -100,12 +94,10 @@ async fn test_uri() {
     use std::convert::TryFrom;
 
     let srv = TestServer::with(|| {
-        fn_service(|io: TcpStream| {
-            async {
-                let mut framed = Framed::new(io, BytesCodec);
-                framed.send(Bytes::from_static(b"test")).await?;
-                Ok::<_, io::Error>(())
-            }
+        fn_service(|io: TcpStream| async {
+            let mut framed = Framed::new(io, BytesCodec);
+            framed.send(Bytes::from_static(b"test")).await?;
+            Ok::<_, io::Error>(())
         })
     });
 
@@ -121,12 +113,10 @@ async fn test_rustls_uri() {
     use std::convert::TryFrom;
 
     let srv = TestServer::with(|| {
-        fn_service(|io: TcpStream| {
-            async {
-                let mut framed = Framed::new(io, BytesCodec);
-                framed.send(Bytes::from_static(b"test")).await?;
-                Ok::<_, io::Error>(())
-            }
+        fn_service(|io: TcpStream| async {
+            let mut framed = Framed::new(io, BytesCodec);
+            framed.send(Bytes::from_static(b"test")).await?;
+            Ok::<_, io::Error>(())
         })
     });
 

actix-ioframe/Cargo.toml

@@ -22,10 +22,12 @@ actix-utils = "1.0.4"
 actix-rt = "1.0.0"
 bytes = "0.5.3"
 either = "1.5.3"
-futures = "0.3.1"
+futures-sink = { version = "0.3.4", default-features = false }
+futures-core = { version = "0.3.4", default-features = false }
 pin-project = "0.4.6"
 log = "0.4"
 
 [dev-dependencies]
-actix-connect = "1.0.0"
+actix-connect = "2.0.0-alpha.2"
 actix-testing = "1.0.0"
+futures-util = { version = "0.3.4", default-features = false }

actix-ioframe/src/connect.rs

@@ -4,7 +4,7 @@ use std::task::{Context, Poll};
 
 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
 use actix_utils::mpsc::Receiver;
-use futures::Stream;
+use futures_core::stream::Stream;
 
 pub struct Connect<Io, Codec>
 where
@@ -42,6 +42,7 @@ where
 pub struct ConnectResult<Io, St, Codec: Encoder + Decoder, Out> {
     pub(crate) state: St,
     pub(crate) out: Option<Out>,
+    #[pin]
     pub(crate) framed: Framed<Io, Codec>,
 }
 
@@ -89,7 +90,7 @@ where
     }
 }
 
-impl<Io, St, Codec, Out> futures::Sink<<Codec as Encoder>::Item>
+impl<Io, St, Codec, Out> futures_sink::Sink<<Codec as Encoder>::Item>
     for ConnectResult<Io, St, Codec, Out>
 where
     Io: AsyncRead + AsyncWrite,
@@ -97,8 +98,8 @@ where
 {
     type Error = <Codec as Encoder>::Error;
 
-    fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        if self.framed.is_write_ready() {
+    fn poll_ready(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        if self.as_mut().project().framed.is_write_ready() {
             Poll::Ready(Ok(()))
         } else {
             Poll::Pending
@@ -112,11 +113,11 @@ where
         self.project().framed.write(item)
     }
 
-    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.get_mut().framed.flush(cx)
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.as_mut().project().framed.flush(cx)
     }
 
-    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.get_mut().framed.close(cx)
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.as_mut().project().framed.close(cx)
     }
 }

actix-ioframe/src/dispatcher.rs

@@ -5,7 +5,8 @@ use std::task::{Context, Poll};
 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
 use actix_service::Service;
 use actix_utils::mpsc;
-use futures::Stream;
+use futures_core::stream::Stream;
+use pin_project::pin_project;
 use log::debug;
 
 use crate::error::ServiceError;
@@ -15,6 +16,7 @@ type Response<U> = <U as Encoder>::Item;
 
 /// FramedTransport - is a future that reads frames from Framed object
 /// and pass then to the service.
+#[pin_project]
 pub(crate) struct Dispatcher<S, T, U, Out>
 where
     S: Service<Request = Request<U>, Response = Option<Response<U>>>,
@@ -29,6 +31,7 @@ where
     service: S,
     sink: Option<Out>,
     state: FramedState<S, U>,
+    #[pin]
     framed: Framed<T, U>,
     rx: mpsc::Receiver<Result<<U as Encoder>::Item, S::Error>>,
 }
@@ -90,26 +93,27 @@ where
     <U as Encoder>::Error: std::fmt::Debug,
     Out: Stream<Item = <U as Encoder>::Item> + Unpin,
 {
-    fn poll_read(&mut self, cx: &mut Context<'_>) -> bool {
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool {
         loop {
-            match self.service.poll_ready(cx) {
+            let this = self.as_mut().project();
+            match this.service.poll_ready(cx) {
                 Poll::Ready(Ok(_)) => {
-                    let item = match self.framed.next_item(cx) {
+                    let item = match this.framed.next_item(cx) {
                         Poll::Ready(Some(Ok(el))) => el,
                         Poll::Ready(Some(Err(err))) => {
-                            self.state = FramedState::FramedError(ServiceError::Decoder(err));
+                            *this.state = FramedState::FramedError(ServiceError::Decoder(err));
                             return true;
                         }
                         Poll::Pending => return false,
                         Poll::Ready(None) => {
                             log::trace!("Client disconnected");
-                            self.state = FramedState::Stopping;
+                            *this.state = FramedState::Stopping;
                             return true;
                         }
                     };
 
-                    let tx = self.rx.sender();
-                    let fut = self.service.call(item);
+                    let tx = this.rx.sender();
+                    let fut = this.service.call(item);
                     actix_rt::spawn(async move {
                         let item = fut.await;
                         let item = match item {
@@ -122,7 +126,7 @@ where
                 }
                 Poll::Pending => return false,
                 Poll::Ready(Err(err)) => {
-                    self.state = FramedState::Error(ServiceError::Service(err));
+                    *this.state = FramedState::Error(ServiceError::Service(err));
                     return true;
                 }
             }
@@ -130,37 +134,38 @@ where
     }
 
     /// write to framed object
-    fn poll_write(&mut self, cx: &mut Context<'_>) -> bool {
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool {
         loop {
-            while !self.framed.is_write_buf_full() {
-                match Pin::new(&mut self.rx).poll_next(cx) {
+            let mut this = self.as_mut().project();
+            while !this.framed.is_write_buf_full() {
+                match Pin::new(&mut this.rx).poll_next(cx) {
                     Poll::Ready(Some(Ok(msg))) => {
-                        if let Err(err) = self.framed.write(msg) {
-                            self.state = FramedState::FramedError(ServiceError::Encoder(err));
+                        if let Err(err) = this.framed.as_mut().write(msg) {
+                            *this.state = FramedState::FramedError(ServiceError::Encoder(err));
                             return true;
                         }
                         continue;
                     }
                     Poll::Ready(Some(Err(err))) => {
-                        self.state = FramedState::Error(ServiceError::Service(err));
+                        *this.state = FramedState::Error(ServiceError::Service(err));
                         return true;
                     }
                     Poll::Ready(None) | Poll::Pending => (),
                 }
 
-                if self.sink.is_some() {
-                    match Pin::new(self.sink.as_mut().unwrap()).poll_next(cx) {
+                if this.sink.is_some() {
+                    match Pin::new(this.sink.as_mut().unwrap()).poll_next(cx) {
                         Poll::Ready(Some(msg)) => {
-                            if let Err(err) = self.framed.write(msg) {
-                                self.state =
+                            if let Err(err) = this.framed.as_mut().write(msg) {
+                                *this.state =
                                     FramedState::FramedError(ServiceError::Encoder(err));
                                 return true;
                             }
                             continue;
                         }
                         Poll::Ready(None) => {
-                            let _ = self.sink.take();
-                            self.state = FramedState::FlushAndStop;
+                            let _ = this.sink.take();
+                            *this.state = FramedState::FlushAndStop;
                             return true;
                         }
                         Poll::Pending => (),
@@ -169,13 +174,13 @@ where
                 break;
             }
 
-            if !self.framed.is_write_buf_empty() {
-                match self.framed.flush(cx) {
+            if !this.framed.is_write_buf_empty() {
+                match this.framed.as_mut().flush(cx) {
                     Poll::Pending => break,
                     Poll::Ready(Ok(_)) => (),
                     Poll::Ready(Err(err)) => {
                         debug!("Error sending data: {:?}", err);
-                        self.state = FramedState::FramedError(ServiceError::Encoder(err));
+                        *this.state = FramedState::FramedError(ServiceError::Encoder(err));
                         return true;
                     }
                 }
@@ -187,13 +192,14 @@ where
     }
 
     pub(crate) fn poll(
-        &mut self,
+        mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), ServiceError<S::Error, U>>> {
-        match self.state {
+        let mut this = self.as_mut().project();
+        match this.state {
             FramedState::Processing => loop {
-                let read = self.poll_read(cx);
-                let write = self.poll_write(cx);
+                let read = self.as_mut().poll_read(cx);
+                let write = self.as_mut().poll_write(cx);
                 if read || write {
                     continue;
                 } else {
@@ -202,18 +208,18 @@ where
             },
             FramedState::Error(_) => {
                 // flush write buffer
-                if !self.framed.is_write_buf_empty() {
-                    if let Poll::Pending = self.framed.flush(cx) {
+                if !this.framed.is_write_buf_empty() {
+                    if let Poll::Pending = this.framed.flush(cx) {
                         return Poll::Pending;
                     }
                 }
-                Poll::Ready(Err(self.state.take_error()))
+                Poll::Ready(Err(this.state.take_error()))
             }
             FramedState::FlushAndStop => {
                 // drain service responses
-                match Pin::new(&mut self.rx).poll_next(cx) {
+                match Pin::new(this.rx).poll_next(cx) {
                     Poll::Ready(Some(Ok(msg))) => {
-                        if let Err(_) = self.framed.write(msg) {
+                        if this.framed.as_mut().write(msg).is_err() {
                             return Poll::Ready(Ok(()));
                         }
                     }
@@ -222,8 +228,8 @@ where
                 }
 
                 // flush io
-                if !self.framed.is_write_buf_empty() {
-                    match self.framed.flush(cx) {
+                if !this.framed.is_write_buf_empty() {
+                    match this.framed.flush(cx) {
                         Poll::Ready(Err(err)) => {
                             debug!("Error sending data: {:?}", err);
                         }
@@ -235,7 +241,7 @@ where
                 };
                 Poll::Ready(Ok(()))
             }
-            FramedState::FramedError(_) => Poll::Ready(Err(self.state.take_framed_error())),
+            FramedState::FramedError(_) => Poll::Ready(Err(this.state.take_framed_error())),
             FramedState::Stopping => Poll::Ready(Ok(())),
         }
     }

actix-ioframe/src/service.rs

@@ -7,7 +7,7 @@ use std::task::{Context, Poll};
 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
 use actix_service::{IntoService, IntoServiceFactory, Service, ServiceFactory};
 use either::Either;
-use futures::{ready, Stream};
+use futures_core::{ready, stream::Stream};
 use pin_project::project;
 
 use crate::connect::{Connect, ConnectResult};
@@ -357,7 +357,7 @@ where
 {
     Connect(#[pin] C::Future, Rc<T>),
     Handler(#[pin] T::Future, Option<Framed<Io, Codec>>, Option<Out>),
-    Dispatcher(Dispatcher<T::Service, Io, Codec, Out>),
+    Dispatcher(#[pin] Dispatcher<T::Service, Io, Codec, Out>),
 }
 
 impl<St, Io, Codec, Out, C, T> FramedServiceImplResponseInner<St, Io, Codec, Out, C, T>
@@ -408,7 +408,7 @@ where
                     Poll::Ready(Err(e)) => Either::Right(Poll::Ready(Err(e.into()))),
                 }
             }
-            FramedServiceImplResponseInner::Dispatcher(ref mut fut) => {
+            FramedServiceImplResponseInner::Dispatcher(fut) => {
                 Either::Right(fut.poll(cx))
             }
         }

actix-ioframe/tests/test_server.rs

@@ -6,7 +6,7 @@ use actix_service::{fn_factory_with_config, fn_service, IntoService, Service};
 use actix_testing::TestServer;
 use actix_utils::mpsc;
 use bytes::{Bytes, BytesMut};
-use futures::future::ok;
+use futures_util::future::ok;
 
 use actix_ioframe::{Builder, Connect, FactoryBuilder};
 

actix-macros/.gitignore

@@ -0,0 +1 @@
+/wip

actix-macros/CHANGES.md

@@ -0,0 +1,9 @@
+# CHANGES
+
+## 0.1.2 - 2020-05-18
+
+### Changed
+
+* Forward actix_rt::test arguments to test function [#127]
+
+[#127]: https://github.com/actix/actix-net/pull/127

actix-macros/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-macros"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix runtime macros"
 repository = "https://github.com/actix/actix-net"
@@ -8,14 +8,16 @@ documentation = "https://docs.rs/actix-macros/"
 categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
 edition = "2018"
-workspace = ".."
 
 [lib]
 proc-macro = true
 
 [dependencies]
-quote = "^1"
+quote = "1.0.3"
 syn = { version = "^1", features = ["full"] }
 
 [dev-dependencies]
-actix-rt = { version = "1.0.0" }
+actix-rt = "1.0"
+
+futures-util = { version = "0.3", default-features = false }
+trybuild = "1"

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 {
@@ -54,12 +55,11 @@ pub fn main(_: TokenStream, item: TokenStream) -> TokenStream {
 /// ```
 #[proc_macro_attribute]
 pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
-    let input = syn::parse_macro_input!(item as syn::ItemFn);
-
-    let ret = &input.sig.output;
-    let name = &input.sig.ident;
-    let body = &input.block;
+    let mut input = syn::parse_macro_input!(item as syn::ItemFn);
     let attrs = &input.attrs;
+    let vis = &input.vis;
+    let sig = &mut input.sig;
+    let body = &input.block;
     let mut has_test_attr = false;
 
     for attr in attrs {
@@ -68,7 +68,7 @@ pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
         }
     }
 
-    if input.sig.asyncness.is_none() {
+    if sig.asyncness.is_none() {
         return syn::Error::new_spanned(
             input.sig.fn_token,
             format!("only async fn is supported, {}", input.sig.ident),
@@ -77,10 +77,12 @@ pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
         .into();
     }
 
+    sig.asyncness = None;
+
     let result = if has_test_attr {
         quote! {
             #(#attrs)*
-            fn #name() #ret {
+            #vis #sig {
                 actix_rt::System::new("test")
                     .block_on(async { #body })
             }
@@ -89,7 +91,7 @@ pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
         quote! {
             #[test]
             #(#attrs)*
-            fn #name() #ret {
+            #vis #sig {
                 actix_rt::System::new("test")
                     .block_on(async { #body })
             }

actix-macros/tests/trybuild.rs

@@ -0,0 +1,9 @@
+#[test]
+fn compile_macros() {
+    let t = trybuild::TestCases::new();
+    t.pass("tests/trybuild/main-01-basic.rs");
+    t.compile_fail("tests/trybuild/main-02-only-async.rs");
+
+    t.pass("tests/trybuild/test-01-basic.rs");
+    t.pass("tests/trybuild/test-02-keep-attrs.rs");
+}

actix-macros/tests/trybuild/main-01-basic.rs

@@ -0,0 +1,4 @@
+#[actix_rt::main]
+async fn main() {
+    println!("Hello world");
+}

actix-macros/tests/trybuild/main-02-only-async.rs

@@ -0,0 +1,4 @@
+#[actix_rt::main]
+fn main() {
+    futures_util::future::ready(()).await
+}

actix-macros/tests/trybuild/main-02-only-async.stderr

@@ -0,0 +1,14 @@
+error: only async fn is supported
+ --> $DIR/main-02-only-async.rs:2:1
+  |
+2 | fn main() {
+  | ^^
+
+error[E0601]: `main` function not found in crate `$CRATE`
+ --> $DIR/main-02-only-async.rs:1:1
+  |
+1 | / #[actix_rt::main]
+2 | | fn main() {
+3 | |     futures_util::future::ready(()).await
+4 | | }
+  | |_^ consider adding a `main` function to `$DIR/tests/trybuild/main-02-only-async.rs`

actix-macros/tests/trybuild/test-01-basic.rs

@@ -0,0 +1,6 @@
+#[actix_rt::test]
+async fn my_test() {
+    assert!(true);
+}
+
+fn main() {}

actix-macros/tests/trybuild/test-02-keep-attrs.rs

@@ -0,0 +1,7 @@
+#[actix_rt::test]
+#[should_panic]
+async fn my_test() {
+    todo!()
+}
+
+fn main() {}

actix-rt/CHANGES.md

@@ -1,5 +1,28 @@
 # Changes
 
+## [1.1.1] - 2020-04-30
+
+### Fixed
+
+* Fix memory leak due to [#94] (see [#129] for more detail)
+
+[#129]: https://github.com/actix/actix-net/issues/129
+
+## [1.1.0] - 2020-04-08
+
+**This version has been yanked.**
+
+### Added
+
+* Expose `System::is_set` to check if current system has ben started [#99]
+* Add `Arbiter::is_running` to check if event loop is running [#124]
+* Add `Arbiter::local_join` associated function
+  to get be able to `await` for spawned futures [#94]
+
+[#94]: https://github.com/actix/actix-net/pull/94
+[#99]: https://github.com/actix/actix-net/pull/99
+[#124]: https://github.com/actix/actix-net/pull/124
+
 ## [1.0.0] - 2019-12-11
 
 * Update dependencies

actix-rt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-rt"
-version = "1.0.0"
+version = "1.1.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix runtime"
 keywords = ["network", "framework", "async", "futures"]
@@ -18,6 +18,8 @@ path = "src/lib.rs"
 [dependencies]
 actix-macros = "0.1.0"
 actix-threadpool = "0.3"
-futures = "0.3.1"
+futures-channel = { version = "0.3.4", default-features = false }
+futures-util = { version = "0.3.4", default-features = false, features = ["alloc"] }
 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"] }
+smallvec = "1"
+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,26 @@ 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;
 
+use smallvec::SmallVec;
+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<SmallVec<[JoinHandle<()>; 8]>> = RefCell::new(SmallVec::new());
     static STORAGE: RefCell<HashMap<TypeId, Box<dyn Any>>> = RefCell::new(HashMap::new());
 );
 
@@ -83,6 +90,11 @@ impl Arbiter {
         })
     }
 
+    /// Check if current arbiter is running.
+    pub fn is_running() -> bool {
+        RUNNING.with(|cell| cell.get())
+    }
+
     /// Stop arbiter from continuing it's event loop.
     pub fn stop(&self) {
         let _ = self.sender.unbounded_send(ArbiterCommand::Stop);
@@ -170,13 +182,20 @@ impl Arbiter {
         RUNNING.with(move |cell| {
             if cell.get() {
                 // Spawn the future on running executor
-                tokio::task::spawn_local(future);
+                let len = PENDING.with(move |cell| {
+                    let mut p = cell.borrow_mut();
+                    p.push(tokio::task::spawn_local(future));
+                    p.len()
+                });
+                if len > 7 {
+                    // Before reaching the inline size
+                    tokio::task::spawn_local(CleanupPending);
+                }
             } 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
                 Q.with(move |cell| {
-                    cell.borrow_mut()
-                        .push(unsafe { Pin::new_unchecked(Box::alloc().init(future)) })
+                    cell.borrow_mut().push(Pin::from(Box::alloc().init(future)))
                 });
             }
         });
@@ -294,6 +313,39 @@ 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(SmallVec::new());
+            future::join_all(current).map(|_| ())
+        })
+    }
+}
+
+/// Future used for cleaning-up already finished `JoinHandle`s
+/// from the `PENDING` list so the vector doesn't grow indefinitely
+struct CleanupPending;
+
+impl Future for CleanupPending {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        PENDING.with(move |cell| {
+            let mut pending = cell.borrow_mut();
+            let mut i = 0;
+            while i != pending.len() {
+                if let Poll::Ready(_) = Pin::new(&mut pending[i]).poll(cx) {
+                    pending.remove(i);
+                } else {
+                    i += 1;
+                }
+            }
+        });
+
+        Poll::Ready(())
+    }
 }
 
 struct ArbiterController {
@@ -329,7 +381,15 @@ impl Future for ArbiterController {
                         return Poll::Ready(());
                     }
                     ArbiterCommand::Execute(fut) => {
-                        tokio::task::spawn_local(fut);
+                        let len = PENDING.with(move |cell| {
+                            let mut p = cell.borrow_mut();
+                            p.push(tokio::task::spawn_local(fut));
+                            p.len()
+                        });
+                        if len > 7 {
+                            // Before reaching the inline size
+                            tokio::task::spawn_local(CleanupPending);
+                        }
                     }
                     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 set, i.e., as already been started.
+    pub fn is_set() -> bool {
         CURRENT.with(|cell| cell.borrow().is_some())
     }
 

actix-rt/tests/integration_tests.rs

@@ -0,0 +1,114 @@
+use std::time::{Duration, Instant};
+
+#[test]
+fn start_and_stop() {
+    actix_rt::System::new("start_and_stop").block_on(async move {
+        assert!(
+            actix_rt::Arbiter::is_running(),
+            "System doesn't seem to have started"
+        );
+    });
+    assert!(
+        !actix_rt::Arbiter::is_running(),
+        "System doesn't seem to have stopped"
+    );
+}
+
+#[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,21 @@
 # Changes
 
+## [1.0.3] - 2020-05-19
+
+### Changed
+
+* Replace deprecated `net2` crate with `socket2` [#140]
+
+[#140]: https://github.com/actix/actix-net/pull/140
+
+## [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.3"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix server - General purpose tcp server"
 keywords = ["network", "framework", "async", "futures"]
@@ -9,7 +9,7 @@ repository = "https://github.com/actix/actix-net.git"
 documentation = "https://docs.rs/actix-server/"
 categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
-exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
+exclude = [".gitignore", ".cargo/config"]
 edition = "2018"
 workspace = ".."
 
@@ -29,14 +29,16 @@ actix-utils = "1.0.4"
 log = "0.4"
 num_cpus = "1.11"
 mio = "0.6.19"
-net2 = "0.2"
-futures = "0.3.1"
+socket2 = "0.3"
+futures-channel = { version = "0.3.4", default-features = false }
+futures-util = { version = "0.3.4", default-features = false, features = ["sink"] }
 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]
 bytes = "0.5"
 env_logger = "0.7"
-actix-testing = "1.0.0"
+actix-testing = "1.0.0"

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

@@ -6,14 +6,13 @@ use std::{io, mem, net};
 use actix_rt::net::TcpStream;
 use actix_rt::time::{delay_until, Instant};
 use actix_rt::{spawn, System};
-use futures::channel::mpsc::{unbounded, UnboundedReceiver};
-use futures::channel::oneshot;
-use futures::future::ready;
-use futures::stream::FuturesUnordered;
-use futures::{ready, Future, FutureExt, Stream, StreamExt};
+use futures_channel::mpsc::{unbounded, UnboundedReceiver};
+use futures_channel::oneshot;
+use futures_util::future::ready;
+use futures_util::stream::FuturesUnordered;
+use futures_util::{future::Future, ready, stream::Stream, FutureExt, StreamExt};
 use log::{error, info};
-use net2::TcpBuilder;
-use num_cpus;
+use socket2::{Domain, Protocol, Socket, Type};
 
 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 {
@@ -486,11 +487,13 @@ pub(super) fn bind_addr<S: net::ToSocketAddrs>(
 }
 
 fn create_tcp_listener(addr: net::SocketAddr, backlog: i32) -> io::Result<net::TcpListener> {
-    let builder = match addr {
-        net::SocketAddr::V4(_) => TcpBuilder::new_v4()?,
-        net::SocketAddr::V6(_) => TcpBuilder::new_v6()?,
+    let domain = match addr {
+        net::SocketAddr::V4(_) => Domain::ipv4(),
+        net::SocketAddr::V6(_) => Domain::ipv6(),
     };
-    builder.reuse_address(true)?;
-    builder.bind(addr)?;
-    Ok(builder.listen(backlog)?)
+    let socket = Socket::new(domain, Type::stream(), Some(Protocol::tcp()))?;
+    socket.set_reuse_address(true)?;
+    socket.bind(&addr.into())?;
+    socket.listen(backlog)?;
+    Ok(socket.into_tcp_listener())
 }

actix-server/src/config.rs

@@ -4,7 +4,7 @@ use std::{fmt, io, net};
 use actix_rt::net::TcpStream;
 use actix_service as actix;
 use actix_utils::counter::CounterGuard;
-use futures::future::{ok, Future, FutureExt, LocalBoxFuture};
+use futures_util::future::{ok, Future, FutureExt, LocalBoxFuture};
 use log::error;
 
 use super::builder::bind_addr;
@@ -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);
     }
 }
@@ -218,7 +218,7 @@ impl ServiceRuntime {
         // let name = name.to_owned();
         if let Some(token) = self.names.get(name) {
             self.services.insert(
-                token.clone(),
+                *token,
                 Box::new(ServiceFactory {
                     inner: service.into_factory(),
                 }),

actix-server/src/server.rs

@@ -3,9 +3,9 @@ use std::io;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-use futures::channel::mpsc::UnboundedSender;
-use futures::channel::oneshot;
-use futures::FutureExt;
+use futures_channel::mpsc::UnboundedSender;
+use futures_channel::oneshot;
+use futures_util::FutureExt;
 
 use crate::builder::ServerBuilder;
 use crate::signals::Signal;

actix-server/src/service.rs

@@ -6,8 +6,8 @@ use std::time::Duration;
 use actix_rt::spawn;
 use actix_service::{self as actix, Service, ServiceFactory as ActixServiceFactory};
 use actix_utils::counter::CounterGuard;
-use futures::future::{err, ok, LocalBoxFuture, Ready};
-use futures::{FutureExt, TryFutureExt};
+use futures_util::future::{err, ok, LocalBoxFuture, Ready};
+use futures_util::{FutureExt, TryFutureExt};
 use log::error;
 
 use super::Token;

actix-server/src/signals.rs

@@ -3,7 +3,7 @@ use std::io;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-use futures::future::lazy;
+use futures_util::future::lazy;
 
 use crate::server::Server;
 

actix-server/src/worker.rs

@@ -7,10 +7,10 @@ use std::time;
 use actix_rt::time::{delay_until, Delay, Instant};
 use actix_rt::{spawn, Arbiter};
 use actix_utils::counter::Counter;
-use futures::channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
-use futures::channel::oneshot;
-use futures::future::{join_all, LocalBoxFuture, MapOk};
-use futures::{Future, FutureExt, Stream, TryFutureExt};
+use futures_channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
+use futures_channel::oneshot;
+use futures_util::future::{join_all, LocalBoxFuture, MapOk};
+use futures_util::{future::Future, stream::Stream, FutureExt, TryFutureExt};
 use log::{error, info, trace};
 
 use crate::accept::AcceptNotify;
@@ -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,23 +1,18 @@
-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;
+use futures_util::future::{lazy, ok};
+use socket2::{Domain, Protocol, Socket, Type};
 
 fn unused_addr() -> net::SocketAddr {
     let addr: net::SocketAddr = "127.0.0.1:0".parse().unwrap();
-    let socket = TcpBuilder::new_v4().unwrap();
-    socket.bind(&addr).unwrap();
-    socket.reuse_address(true).unwrap();
-    let tcp = socket.to_tcp_listener().unwrap();
+    let socket = Socket::new(Domain::ipv4(), Type::stream(), Some(Protocol::tcp())).unwrap();
+    socket.bind(&addr.into()).unwrap();
+    socket.set_reuse_address(true).unwrap();
+    let tcp = socket.into_tcp_listener();
     tcp.local_addr().unwrap()
 }
 
@@ -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_util::sink::SinkExt;
+    use std::io::Read;
+
     let addr = unused_addr();
     let (tx, rx) = mpsc::channel();
 
@@ -82,12 +83,10 @@ fn test_start() {
             .backlog(100)
             .disable_signals()
             .bind("test", addr, move || {
-                fn_service(|io: TcpStream| {
-                    async move {
-                        let mut f = Framed::new(io, BytesCodec);
-                        f.send(Bytes::from_static(b"test")).await.unwrap();
-                        Ok::<_, ()>(())
-                    }
+                fn_service(|io: TcpStream| async move {
+                    let mut f = Framed::new(io, BytesCodec);
+                    f.send(Bytes::from_static(b"test")).await.unwrap();
+                    Ok::<_, ()>(())
                 })
             })
             .unwrap()
@@ -130,7 +129,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 +177,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/Cargo.toml

@@ -11,10 +11,6 @@ categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
 edition = "2018"
 
-[badges]
-travis-ci = { repository = "actix/actix-service", branch = "master" }
-codecov = { repository = "actix/actix-service", branch = "master", service = "github" }
-
 [lib]
 name = "actix_service"
 path = "src/lib.rs"
@@ -25,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

@@ -313,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

@@ -290,6 +290,7 @@ mod tests {
             Poll::Ready(Ok(()))
         }
 
+        #[allow(clippy::unit_arg)]
         fn call(&mut self, req: Self::Request) -> Self::Future {
             ok(req)
         }
@@ -297,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))
             });
@@ -311,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/then.rs

@@ -305,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]
@@ -321,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/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [Unreleased]
+
+* Replace deprecated `net2` crate with `socket2`
+
 ## [1.0.0] - 2019-12-11
 
 * Update actix-server to 1.0.0

actix-testing/Cargo.toml

@@ -23,5 +23,4 @@ actix-server = "1.0.0"
 actix-service = "1.0.0"
 
 log = "0.4"
-net2 = "0.2"
-futures = "0.3.1"
+socket2 = "0.3"

actix-testing/src/lib.rs

@@ -1,13 +1,13 @@
 //! 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};
 
 use actix_rt::{net::TcpStream, System};
 use actix_server::{Server, ServerBuilder, ServiceFactory};
-use net2::TcpBuilder;
+use socket2::{Domain, Protocol, Socket, Type};
 
 #[cfg(not(test))] // Work around for rust-lang/rust#62127
 pub use actix_macros::test;
@@ -110,10 +110,10 @@ impl TestServer {
     /// Get firat available unused local address
     pub fn unused_addr() -> net::SocketAddr {
         let addr: net::SocketAddr = "127.0.0.1:0".parse().unwrap();
-        let socket = TcpBuilder::new_v4().unwrap();
-        socket.bind(&addr).unwrap();
-        socket.reuse_address(true).unwrap();
-        let tcp = socket.to_tcp_listener().unwrap();
+        let socket = Socket::new(Domain::ipv4(), Type::stream(), Some(Protocol::tcp())).unwrap();
+        socket.bind(&addr.into()).unwrap();
+        socket.set_reuse_address(true).unwrap();
+        let tcp = socket.into_tcp_listener();
         tcp.local_addr().unwrap()
     }
 }

actix-threadpool/src/lib.rs

@@ -48,6 +48,8 @@ pub enum BlockingError<E: fmt::Debug> {
     Canceled,
 }
 
+impl<E: fmt::Debug> std::error::Error for BlockingError<E> {}
+
 /// Execute blocking function on a thread pool, returns future that resolves
 /// to result of the function execution.
 pub fn run<F, I, E>(f: F) -> CpuFuture<I, E>

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"]
@@ -38,7 +38,7 @@ actix-utils = "1.0.0"
 actix-rt = "1.0.0"
 derive_more = "0.99.2"
 either = "1.5.2"
-futures = "0.3.1"
+futures-util = { version = "0.3.4", default-features = false }
 log = "0.4"
 
 # openssl
@@ -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-tls/src/nativetls.rs

@@ -4,7 +4,7 @@ use std::task::{Context, Poll};
 use actix_codec::{AsyncRead, AsyncWrite};
 use actix_service::{Service, ServiceFactory};
 use actix_utils::counter::Counter;
-use futures::future::{self, FutureExt, LocalBoxFuture, TryFutureExt};
+use futures_util::future::{self, FutureExt, LocalBoxFuture, TryFutureExt};
 pub use native_tls::Error;
 pub use tokio_tls::{TlsAcceptor, TlsStream};
 

actix-tls/src/openssl.rs

@@ -9,7 +9,7 @@ pub use tokio_openssl::{HandshakeError, SslStream};
 use actix_codec::{AsyncRead, AsyncWrite};
 use actix_service::{Service, ServiceFactory};
 use actix_utils::counter::{Counter, CounterGuard};
-use futures::future::{ok, FutureExt, LocalBoxFuture, Ready};
+use futures_util::future::{ok, FutureExt, LocalBoxFuture, Ready};
 
 use crate::MAX_CONN_COUNTER;
 
@@ -105,7 +105,7 @@ impl<T: AsyncRead + AsyncWrite + Unpin> Future for AcceptorServiceResponse<T> {
     type Output = Result<SslStream<T>, HandshakeError<T>>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let io = futures::ready!(Pin::new(&mut self.fut).poll(cx))?;
+        let io = futures_util::ready!(Pin::new(&mut self.fut).poll(cx))?;
         Poll::Ready(Ok(io))
     }
 }

actix-tls/src/rustls.rs

@@ -8,7 +8,7 @@ use std::task::{Context, Poll};
 use actix_codec::{AsyncRead, AsyncWrite};
 use actix_service::{Service, ServiceFactory};
 use actix_utils::counter::{Counter, CounterGuard};
-use futures::future::{ok, Ready};
+use futures_util::future::{ok, Ready};
 use tokio_rustls::{Accept, TlsAcceptor};
 
 pub use rust_tls::{ServerConfig, Session};
@@ -108,7 +108,7 @@ impl<T: AsyncRead + AsyncWrite + Unpin> Future for AcceptorServiceFut<T> {
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.get_mut();
 
-        let res = futures::ready!(Pin::new(&mut this.fut).poll(cx));
+        let res = futures_util::ready!(Pin::new(&mut this.fut).poll(cx));
         match res {
             Ok(io) => Poll::Ready(Ok(io)),
             Err(e) => Poll::Ready(Err(e)),

actix-tracing/Cargo.toml

@@ -17,7 +17,7 @@ path = "src/lib.rs"
 
 [dependencies]
 actix-service = "1.0.4"
-futures-util = "0.3.1"
+futures-util = { version = "0.3.4", default-features = false }
 tracing = "0.1"
 tracing-futures = "0.2"
 

actix-utils/Cargo.toml

@@ -22,7 +22,9 @@ actix-codec = "0.2.0"
 bitflags = "1.2"
 bytes = "0.5.3"
 either = "1.5.3"
-futures = "0.3.1"
+futures-channel = { version = "0.3.4", default-features = false }
+futures-sink = { version = "0.3.4", default-features = false }
+futures-util = { version = "0.3.4", default-features = false }
 pin-project = "0.4.6"
 log = "0.4"
 slab = "0.4"

actix-utils/src/condition.rs

@@ -96,7 +96,7 @@ impl Drop for Waiter {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use futures::future::lazy;
+    use futures_util::future::lazy;
 
     #[actix_rt::test]
     async fn test_condition() {
@@ -107,7 +107,7 @@ mod tests {
             Poll::Pending
         );
         cond.notify();
-        assert_eq!(waiter.await, ());
+        waiter.await;
 
         let mut waiter = cond.wait();
         assert_eq!(
@@ -121,7 +121,7 @@ mod tests {
         );
 
         drop(cond);
-        assert_eq!(waiter.await, ());
-        assert_eq!(waiter2.await, ());
+        waiter.await;
+        waiter2.await;
     }
 }

actix-utils/src/either.rs

@@ -3,7 +3,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use actix_service::{Service, ServiceFactory};
-use futures::{future, ready, Future};
+use futures_util::{future, ready, future::Future};
 
 /// Combine two different service types into a single type.
 ///

actix-utils/src/framed.rs

@@ -6,7 +6,7 @@ use std::{fmt, mem};
 
 use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed};
 use actix_service::{IntoService, Service};
-use futures::{Future, FutureExt, Stream};
+use futures_util::{future::Future, FutureExt, stream::Stream};
 use log::debug;
 
 use crate::mpsc;
@@ -77,6 +77,7 @@ where
 {
     service: S,
     state: State<S, U>,
+    #[pin]
     framed: Framed<T, U>,
     rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
     tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>>,
@@ -169,7 +170,7 @@ where
         &mut self.framed
     }
 
-    fn poll_read(&mut self, cx: &mut Context<'_>) -> bool
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
     where
         S: Service<Request = Request<U>, Response = Response<U>>,
         S::Error: 'static,
@@ -180,29 +181,30 @@ where
         <U as Encoder>::Error: std::fmt::Debug,
     {
         loop {
-            match self.service.poll_ready(cx) {
+            let this = self.as_mut().project();
+            match this.service.poll_ready(cx) {
                 Poll::Ready(Ok(_)) => {
-                    let item = match self.framed.next_item(cx) {
+                    let item = match this.framed.next_item(cx) {
                         Poll::Ready(Some(Ok(el))) => el,
                         Poll::Ready(Some(Err(err))) => {
-                            self.state = State::FramedError(DispatcherError::Decoder(err));
+                            *this.state = State::FramedError(DispatcherError::Decoder(err));
                             return true;
                         }
                         Poll::Pending => return false,
                         Poll::Ready(None) => {
-                            self.state = State::Stopping;
+                            *this.state = State::Stopping;
                             return true;
                         }
                     };
 
-                    let tx = self.tx.clone();
-                    actix_rt::spawn(self.service.call(item).map(move |item| {
+                    let tx = this.tx.clone();
+                    actix_rt::spawn(this.service.call(item).map(move |item| {
                         let _ = tx.send(item.map(Message::Item));
                     }));
                 }
                 Poll::Pending => return false,
                 Poll::Ready(Err(err)) => {
-                    self.state = State::Error(DispatcherError::Service(err));
+                    *this.state = State::Error(DispatcherError::Service(err));
                     return true;
                 }
             }
@@ -210,7 +212,7 @@ where
     }
 
     /// write to framed object
-    fn poll_write(&mut self, cx: &mut Context<'_>) -> bool
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> bool
     where
         S: Service<Request = Request<U>, Response = Response<U>>,
         S::Error: 'static,
@@ -221,33 +223,34 @@ where
         <U as Encoder>::Error: std::fmt::Debug,
     {
         loop {
-            while !self.framed.is_write_buf_full() {
-                match Pin::new(&mut self.rx).poll_next(cx) {
+            let mut this = self.as_mut().project();
+            while !this.framed.is_write_buf_full() {
+                match Pin::new(&mut this.rx).poll_next(cx) {
                     Poll::Ready(Some(Ok(Message::Item(msg)))) => {
-                        if let Err(err) = self.framed.write(msg) {
-                            self.state = State::FramedError(DispatcherError::Encoder(err));
+                        if let Err(err) = this.framed.as_mut().write(msg) {
+                            *this.state = State::FramedError(DispatcherError::Encoder(err));
                             return true;
                         }
                     }
                     Poll::Ready(Some(Ok(Message::Close))) => {
-                        self.state = State::FlushAndStop;
+                        *this.state = State::FlushAndStop;
                         return true;
                     }
                     Poll::Ready(Some(Err(err))) => {
-                        self.state = State::Error(DispatcherError::Service(err));
+                        *this.state = State::Error(DispatcherError::Service(err));
                         return true;
                     }
                     Poll::Ready(None) | Poll::Pending => break,
                 }
             }
 
-            if !self.framed.is_write_buf_empty() {
-                match self.framed.flush(cx) {
+            if !this.framed.is_write_buf_empty() {
+                match this.framed.flush(cx) {
                     Poll::Pending => break,
                     Poll::Ready(Ok(_)) => (),
                     Poll::Ready(Err(err)) => {
                         debug!("Error sending data: {:?}", err);
-                        self.state = State::FramedError(DispatcherError::Encoder(err));
+                        *this.state = State::FramedError(DispatcherError::Encoder(err));
                         return true;
                     }
                 }
@@ -279,7 +282,7 @@ where
 
             return match this.state {
                 State::Processing => {
-                    if self.poll_read(cx) || self.poll_write(cx) {
+                    if self.as_mut().poll_read(cx) || self.as_mut().poll_write(cx) {
                         continue;
                     } else {
                         Poll::Pending
@@ -287,12 +290,12 @@ where
                 }
                 State::Error(_) => {
                     // flush write buffer
-                    if !self.framed.is_write_buf_empty() {
-                        if let Poll::Pending = self.framed.flush(cx) {
+                    if !this.framed.is_write_buf_empty() {
+                        if let Poll::Pending = this.framed.flush(cx) {
                             return Poll::Pending;
                         }
                     }
-                    Poll::Ready(Err(self.state.take_error()))
+                    Poll::Ready(Err(this.state.take_error()))
                 }
                 State::FlushAndStop => {
                     if !this.framed.is_write_buf_empty() {

actix-utils/src/inflight.rs

@@ -4,7 +4,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use actix_service::{IntoService, Service, Transform};
-use futures::future::{ok, Ready};
+use futures_util::future::{ok, Ready};
 
 use super::counter::{Counter, CounterGuard};
 
@@ -116,7 +116,7 @@ mod tests {
 
     use super::*;
     use actix_service::{apply, fn_factory, Service, ServiceFactory};
-    use futures::future::{lazy, ok, FutureExt, LocalBoxFuture};
+    use futures_util::future::{lazy, ok, FutureExt, LocalBoxFuture};
 
     struct SleepService(Duration);
 

actix-utils/src/keepalive.rs

@@ -7,7 +7,7 @@ use std::time::Duration;
 
 use actix_rt::time::{delay_until, Delay, Instant};
 use actix_service::{Service, ServiceFactory};
-use futures::future::{ok, Ready};
+use futures_util::future::{ok, Ready};
 
 use super::time::{LowResTime, LowResTimeService};
 

actix-utils/src/mpsc.rs

@@ -6,7 +6,8 @@ use std::fmt;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-use futures::{Sink, Stream};
+use futures_sink::Sink;
+use futures_util::stream::Stream;
 
 use crate::cell::Cell;
 use crate::task::LocalWaker;
@@ -180,8 +181,8 @@ impl<T> SendError<T> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use futures::future::lazy;
-    use futures::{Stream, StreamExt};
+    use futures_util::future::lazy;
+    use futures_util::{stream::Stream, StreamExt};
 
     #[actix_rt::test]
     async fn test_mpsc() {

actix-utils/src/oneshot.rs

@@ -3,7 +3,7 @@ use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-pub use futures::channel::oneshot::Canceled;
+pub use futures_channel::oneshot::Canceled;
 use slab::Slab;
 
 use crate::cell::Cell;
@@ -253,7 +253,7 @@ impl<T> Future for PReceiver<T> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use futures::future::lazy;
+    use futures_util::future::lazy;
 
     #[actix_rt::test]
     async fn test_oneshot() {

actix-utils/src/order.rs

@@ -8,7 +8,7 @@ use std::rc::Rc;
 use std::task::{Context, Poll};
 
 use actix_service::{IntoService, Service, Transform};
-use futures::future::{ok, Ready};
+use futures_util::future::{ok, Ready};
 
 use crate::oneshot;
 use crate::task::LocalWaker;
@@ -210,8 +210,8 @@ mod tests {
 
     use super::*;
     use actix_service::Service;
-    use futures::channel::oneshot;
-    use futures::future::{lazy, poll_fn, FutureExt, LocalBoxFuture};
+    use futures_channel::oneshot;
+    use futures_util::future::{lazy, poll_fn, FutureExt, LocalBoxFuture};
 
     struct Srv;
 
@@ -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
                     })

actix-utils/src/stream.rs

@@ -3,7 +3,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use actix_service::{IntoService, Service};
-use futures::{FutureExt, Stream};
+use futures_util::{FutureExt, stream::Stream};
 
 use crate::mpsc;
 

actix-utils/src/time.rs

@@ -4,7 +4,7 @@ use std::time::{self, Duration, Instant};
 
 use actix_rt::time::delay_for;
 use actix_service::{Service, ServiceFactory};
-use futures::future::{ok, ready, FutureExt, Ready};
+use futures_util::future::{ok, ready, FutureExt, Ready};
 
 use super::cell::Cell;
 

actix-utils/src/timeout.rs

@@ -10,7 +10,7 @@ use std::{fmt, time};
 
 use actix_rt::time::{delay_for, Delay};
 use actix_service::{IntoService, Service, Transform};
-use futures::future::{ok, Ready};
+use futures_util::future::{ok, Ready};
 
 /// Applies a timeout to requests.
 #[derive(Debug)]
@@ -183,7 +183,7 @@ mod tests {
 
     use super::*;
     use actix_service::{apply, fn_factory, Service, ServiceFactory};
-    use futures::future::{ok, FutureExt, LocalBoxFuture};
+    use futures_util::future::{ok, FutureExt, LocalBoxFuture};
 
     struct SleepService(Duration);
 

examples/basic.rs

@@ -11,7 +11,7 @@ use actix_codec::{AsyncRead, AsyncWrite};
 use actix_rt::System;
 use actix_server::{Io, Server};
 use actix_service::{service_fn, NewService};
-use futures::{future, Future};
+use futures_util::{future, Future};
 use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod};
 use tokio_openssl::SslAcceptorExt;
 

examples/ssl.rs

@@ -7,7 +7,7 @@ use std::sync::{
 use actix_rt::System;
 use actix_server::{ssl, Server};
 use actix_service::NewService;
-use futures::future;
+use futures_util::future;
 use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod};
 
 #[derive(Debug)]

router/src/lib.rs

@@ -69,7 +69,7 @@ impl<'a> IntoPattern for &'a str {
     }
 
     fn patterns(&self) -> Vec<String> {
-        vec![self.to_string()]
+        vec![(*self).to_string()]
     }
 }
 
@@ -79,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

@@ -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
@@ -734,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![
@@ -899,31 +900,31 @@ mod tests {
     fn test_resource_path() {
         let mut s = String::new();
         let resource = ResourceDef::new("/user/{item1}/test");
-        assert!(resource.resource_path(&mut s, &mut (&["user1"]).into_iter()));
+        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, &mut (&["item", "item2"]).into_iter()));
+        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, &mut (&["item", "item2"]).into_iter()));
+        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, &mut (&["item", "item2"]).into_iter()));
+        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, &mut (&["item"]).into_iter()));
+        assert!(!resource.resource_path(&mut s, &mut (&["item"]).iter()));
 
         let mut s = String::new();
-        assert!(resource.resource_path(&mut s, &mut (&["item", "item2"]).into_iter()));
+        assert!(resource.resource_path(&mut s, &mut (&["item", "item2"]).iter()));
         assert_eq!(s, "/user/item/item2/");
-        assert!(!resource.resource_path(&mut s, &mut (&["item"]).into_iter()));
+        assert!(!resource.resource_path(&mut s, &mut (&["item"]).iter()));
 
         let mut s = String::new();
         assert!(resource.resource_path(&mut s, &mut vec!["item", "item2"].into_iter()));

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();

router/src/url.rs

@@ -31,7 +31,7 @@ fn set_bit(array: &mut [u8], ch: u8) {
 }
 
 thread_local! {
-    static DEFAULT_QUOTER: Quoter = { Quoter::new(b"@:", b"/+") };
+    static DEFAULT_QUOTER: Quoter = Quoter::new(b"@:", b"/+");
 }
 
 #[derive(Default, Clone, Debug)]

string/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.1.5] - 2020-03-30
+
+* Serde support
+
 ## [0.1.4] - 2020-01-14
 
 * Fix `AsRef<str>` impl

string/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bytestring"
-version = "0.1.4"
+version = "0.1.5"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "A UTF-8 encoded string with Bytes as a storage"
 keywords = ["actix"]
@@ -16,3 +16,7 @@ path = "src/lib.rs"
 
 [dependencies]
 bytes = "0.5.3"
+serde = { version = "1.0", optional = true }
+
+[dev-dependencies]
+serde_json = "1.0"

string/src/lib.rs

@@ -1,10 +1,11 @@
-//! A utl-8 encoded read-only string with Bytes as a storage.
+//! A UTF-8 encoded read-only string using Bytes as storage.
+
 use std::convert::TryFrom;
 use std::{borrow, fmt, hash, ops, str};
 
 use bytes::Bytes;
 
-/// A utf-8 encoded string with [`Bytes`] as a storage.
+/// 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, Ord, PartialOrd, Default)]
@@ -159,6 +160,34 @@ impl fmt::Display for ByteString {
     }
 }
 
+#[cfg(feature = "serde")]
+mod serde {
+    use serde::de::{Deserialize, Deserializer};
+    use serde::ser::{Serialize, Serializer};
+
+    use super::ByteString;
+
+    impl Serialize for ByteString {
+        #[inline]
+        fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+        where
+            S: Serializer,
+        {
+            serializer.serialize_str(self.as_ref())
+        }
+    }
+
+    impl<'de> Deserialize<'de> for ByteString {
+        #[inline]
+        fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+        where
+            D: Deserializer<'de>,
+        {
+            String::deserialize(deserializer).map(ByteString::from)
+        }
+    }
+}
+
 #[cfg(test)]
 mod test {
     use super::*;
@@ -204,7 +233,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");
     }
 
@@ -222,4 +251,18 @@ mod test {
     fn test_try_from_bytesmut() {
         let _ = ByteString::try_from(bytes::BytesMut::from(&b"nice bytes"[..])).unwrap();
     }
+
+    #[cfg(feature = "serde")]
+    #[test]
+    fn test_serialize() {
+        let s: ByteString = serde_json::from_str(r#""nice bytes""#).unwrap();
+        assert_eq!(s, "nice bytes");
+    }
+
+    #[cfg(feature = "serde")]
+    #[test]
+    fn test_deserialize() {
+        let s = serde_json::to_string(&ByteString::from_static("nice bytes")).unwrap();
+        assert_eq!(s, r#""nice bytes""#);
+    }
 }