chore: prepare tokio-macros v0.3.2 (#3295 )

chore: fix stress test (#3296 )
chore: prepare v0.3.6 release (#3276 )
2026-09-09 00:00:08 +02:00 · 2020-12-19 17:03:42 +01:00 · 2020-12-19 12:11:22 +01:00 · 2020-12-14 13:51:38 -08:00 · 2020-12-10 12:00:50 +01:00 · 2020-12-10 11:30:09 +01:00
520 changed files with 8772 additions and 44441 deletions
@@ -1,8 +0,0 @@
-# See https://github.com/rustsec/rustsec/blob/59e1d2ad0b9cbc6892c26de233d4925074b4b97b/cargo-audit/audit.toml.example for example.
-
-[advisories]
-ignore = [
-    # We depend on nix 0.22 only via mio-aio, a dev-dependency.
-    # https://github.com/tokio-rs/tokio/pull/4255#issuecomment-974786349
-    "RUSTSEC-2021-0119",
-]
@@ -1,25 +0,0 @@
-version: 2.1        
-jobs:
-  test-arm:
-    machine:
-      image: ubuntu-2004:202101-01
-    resource_class: arm.medium
-    environment:
-      # Change to pin rust versino
-      RUST_STABLE: stable
-    steps:
-      - checkout
-      - run:
-          name: Install Rust
-          command: |
-            curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs -o rustup.sh
-            chmod +x rustup.sh
-            ./rustup.sh -y --default-toolchain $RUST_STABLE
-            source "$HOME"/.cargo/env
-      # Only run Tokio tests
-      - run: cargo test --all-features -p tokio
-
-workflows:
-  ci:
-    jobs:
-      - test-arm
@@ -1,51 +1,19 @@
 freebsd_instance:
-  image: freebsd-12-3-release-amd64
-env:
-  RUST_STABLE: stable
-  RUST_NIGHTLY: nightly-2022-03-21
-  RUSTFLAGS: -D warnings
+  image: freebsd-12-2-release-amd64

 # Test FreeBSD in a full VM on cirrus-ci.com.  Test the i686 target too, in the
 # same VM.  The binary will be built in 32-bit mode, but will execute on a
 # 64-bit kernel and in a 64-bit environment.  Our tests don't execute any of
 # the system's binaries, so the environment shouldn't matter.
 task:
-  name: FreeBSD 64-bit
-  setup_script:
-    - pkg install -y bash curl
-    - curl https://sh.rustup.rs -sSf --output rustup.sh
-    - sh rustup.sh -y --profile minimal --default-toolchain $RUST_STABLE
-    - . $HOME/.cargo/env
-    - |
-      echo "~~~~ rustc --version ~~~~"
-      rustc --version
-  test_script:
-    - . $HOME/.cargo/env
-    - cargo test --all --all-features
-
-task:
-  name: FreeBSD docs
+  name: FreeBSD
  env:
-    RUSTFLAGS: --cfg docsrs --cfg tokio_unstable
-    RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable -Dwarnings
+    LOOM_MAX_PREEMPTIONS: 2
+    RUSTFLAGS: -Dwarnings
  setup_script:
    - pkg install -y bash curl
    - curl https://sh.rustup.rs -sSf --output rustup.sh
-    - sh rustup.sh -y --profile minimal --default-toolchain $RUST_NIGHTLY
-    - . $HOME/.cargo/env
-    - |
-      echo "~~~~ rustc --version ~~~~"
-      rustc --version
-  test_script:
-    - . $HOME/.cargo/env
-    - cargo doc --lib --no-deps --all-features --document-private-items
-
-task:
-  name: FreeBSD 32-bit
-  setup_script:
-    - pkg install -y bash curl
-    - curl https://sh.rustup.rs -sSf --output rustup.sh
-    - sh rustup.sh -y --profile minimal --default-toolchain $RUST_STABLE
+    - sh rustup.sh -y --profile minimal --default-toolchain stable
    - . $HOME/.cargo/env
    - rustup target add i686-unknown-freebsd
    - |
@@ -53,4 +21,9 @@ task:
      rustc --version
  test_script:
    - . $HOME/.cargo/env
-    - cargo test --all --all-features --target i686-unknown-freebsd
+    - cargo test --all
+    - cargo doc --all --no-deps
+  i686_test_script:
+    - . $HOME/.cargo/env
+    - |
+      cargo test --all --target i686-unknown-freebsd
@@ -1 +0,0 @@
-msrv = "1.49"
@@ -1,3 +0,0 @@
-# These are supported funding model platforms
-
-github: [tokio-rs]
@@ -9,7 +9,12 @@ assignees: ''

 **Version**
 List the versions of all `tokio` crates you are using. The easiest way to get
-this information is using `cargo tree` subcommand:
+this information is using `cargo-tree`.
+
+`cargo install cargo-tree`
+(see install here: https://github.com/sfackler/cargo-tree)
+
+Then:

 `cargo tree | grep tokio`

@@ -1,8 +0,0 @@
-
-R-loom:
- tokio/src/sync/*
- tokio/src/sync/**/*
- tokio-util/src/sync/*
- tokio-util/src/sync/**/*
- tokio/src/runtime/*
- tokio/src/runtime/**/*
@@ -0,0 +1,55 @@
+name: Benchmark
+on:
+  push:
+    branches:
+      - master
+
+jobs:
+  benchmark:
+    name: Benchmark
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        bench:
+          - rt_multi_threaded
+          - sync_mpsc
+          - sync_rwlock
+          - sync_semaphore
+    steps:
+      - uses: actions/checkout@v2
+      - name: Install Rust
+        run: rustup update stable
+
+      # Run benchmark with `go test -bench` and stores the output to a file
+      - name: Run benchmark
+        run: cargo bench --bench ${{ matrix.bench }} | tee ../output.txt
+        working-directory: benches
+
+      # Download previous benchmark result from cache (if exists)
+      - name: Download previous benchmark data
+        uses: actions/cache@v1
+        with:
+          path: ./cache
+          key: ${{ runner.os }}-benchmark
+
+      # Run `github-action-benchmark` action
+      - name: Store benchmark result
+        uses: rhysd/github-action-benchmark@v1
+        with:
+          name: ${{ matrix.bench }}
+          # What benchmark tool the output.txt came from
+          tool: 'cargo'
+          # Where the output from the benchmark tool is stored
+          output-file-path: output.txt
+          # # Where the previous data file is stored
+          # external-data-json-path: ./cache/benchmark-data.json
+          # Workflow will fail when an alert happens
+          fail-on-alert: true
+          # GitHub API token to make a commit comment
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          # Enable alert commit comment
+          comment-on-alert: true
+          alert-comment-cc-users: '@tokio-rs/maintainers'
+          auto-push: true
+
+      # Upload the updated cache file for the next job by actions/cache
@@ -1,23 +1,16 @@
 on:
  push:
-    branches: ["master", "tokio-*.x"]
+    branches: ["v0.3.x"]
  pull_request:
-    branches: ["master", "tokio-*.x"]
+    branches: ["v0.3.x"]

 name: CI

 env:
  RUSTFLAGS: -Dwarnings
  RUST_BACKTRACE: 1
-  # Change to specific Rust release to pin
-  rust_stable: stable
-  rust_nightly: nightly-2022-03-21
-  rust_clippy: 1.52.0
-  rust_min: 1.49.0
-
-defaults:
-  run:
-    shell: bash
+  nightly: nightly-2020-09-21
+  minrust: 1.45.2

 jobs:
  # Depends on all action sthat are required for a "successful" CI run.
@@ -27,7 +20,6 @@ jobs:
    needs:
      - test
      - test-unstable
-      - test-parking_lot
      - miri
      - cross
      - features
@@ -35,11 +27,7 @@ jobs:
      - fmt
      - clippy
      - docs
-      - valgrind
-      - loom-compile
-      - check-readme
-      - test-hyper
-      - wasm32-unknown-unknown
+      - loom
    steps:
      - run: exit 0

@@ -54,14 +42,8 @@ jobs:
          - macos-latest
    steps:
      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
-        with:
-            toolchain: ${{ env.rust_stable }}
-            override: true
      - name: Install Rust
        run: rustup update stable
-      - uses: Swatinem/rust-cache@v1
      - name: Install cargo-hack
        run: cargo install cargo-hack

@@ -85,72 +67,6 @@ jobs:
        run: cargo hack test --each-feature
        working-directory: tests-build

-      # Build benchmarks. Run of benchmarks is done by bench.yml workflow.
-      - name: build benches
-        run: cargo build --benches
-        working-directory: benches
-        # bench.yml workflow runs benchmarks only on linux.
-        if: startsWith(matrix.os, 'ubuntu')
-
-  test-parking_lot:
-    # The parking_lot crate has a feature called send_guard which changes when
-    # some of its types are Send. Tokio has some measures in place to prevent
-    # this from affecting when Tokio types are Send, and this test exists to
-    # ensure that those measures are working.
-    #
-    # This relies on the potentially affected Tokio type being listed in
-    # `tokio/tokio/tests/async_send_sync.rs`.
-    name: compile tests with parking lot send guards
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
-        with:
-            toolchain: ${{ env.rust_stable }}
-            override: true
-      - uses: Swatinem/rust-cache@v1
-      - name: Enable parking_lot send_guard feature
-        # Inserts the line "plsend = ["parking_lot/send_guard"]" right after [features]
-        run: sed -i '/\[features\]/a plsend = ["parking_lot/send_guard"]' tokio/Cargo.toml
-      - name: Compile tests with all features enabled
-        run: cargo build --workspace --all-features --tests
-
-  valgrind:
-    name: valgrind
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
-        with:
-            toolchain: ${{ env.rust_stable }}
-            override: true
-      - uses: Swatinem/rust-cache@v1
-
-      - name: Install Valgrind
-        run: |
-          sudo apt-get update -y
-          sudo apt-get install -y valgrind
-
-      # Compile tests
-      - name: cargo build test-mem
-        run: cargo build --features rt-net --bin test-mem
-        working-directory: tests-integration
-
-      # Run with valgrind
-      - name: Run valgrind test-mem
-        run: valgrind --error-exitcode=1 --leak-check=full --show-leak-kinds=all ./target/debug/test-mem
-
-      # Compile tests
-      - name: cargo build test-process-signal
-        run: cargo build --features rt-process-signal --bin test-process-signal
-        working-directory: tests-integration
-
-      # Run with valgrind
-      - name: Run valgrind test-process-signal
-        run: valgrind --error-exitcode=1 --leak-check=full --show-leak-kinds=all ./target/debug/test-process-signal
-
  test-unstable:
    name: test tokio full --unstable
    runs-on: ${{ matrix.os }}
@@ -162,54 +78,44 @@ jobs:
          - macos-latest
    steps:
      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
-        with:
-            toolchain: ${{ env.rust_stable }}
-            override: true
-      - uses: Swatinem/rust-cache@v1
+      - name: Install Rust
+        run: rustup update stable
+
      # Run `tokio` with "unstable" cfg flag.
      - name: test tokio full --cfg unstable
-        run: cargo test --all-features
+        run: cargo test --features full
        working-directory: tokio
        env:
          RUSTFLAGS: --cfg tokio_unstable -Dwarnings
-          # in order to run doctests for unstable features, we must also pass
-          # the unstable cfg to RustDoc
-          RUSTDOCFLAGS: --cfg tokio_unstable

  miri:
    name: miri
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_nightly }}
-        uses: actions-rs/toolchain@v1
+      - uses: actions-rs/toolchain@v1
        with:
-          toolchain: ${{ env.rust_nightly }}
-          components: miri
+          toolchain: ${{ env.nightly }}
          override: true
-      - uses: Swatinem/rust-cache@v1
-      - name: miri
-        # Many of tests in tokio/tests and doctests use #[tokio::test] or
-        # #[tokio::main] that calls epoll_create1 that Miri does not support.
-        run: cargo miri test --features full --lib --no-fail-fast
-        working-directory: tokio
-        env:
-          MIRIFLAGS: -Zmiri-disable-isolation -Zmiri-tag-raw-pointers
-          PROPTEST_CASES: 10
+      - name: Install Miri
+        run: |
+          set -e
+          rustup component add miri
+          cargo miri setup
+          rm -rf tokio/tests

+      - name: miri
+        run: cargo miri test --features rt,rt-multi-thread,sync task
+        working-directory: tokio
  san:
    name: san
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_nightly }}
-        uses: actions-rs/toolchain@v1
+      - uses: actions-rs/toolchain@v1
        with:
-          toolchain: ${{ env.rust_nightly }}
+          toolchain: ${{ env.nightly }}
          override: true
-      - uses: Swatinem/rust-cache@v1
      - name: asan
        run: cargo test --all-features --target x86_64-unknown-linux-gnu --lib -- --test-threads 1
        working-directory: tokio
@@ -230,13 +136,11 @@ jobs:
          - arm-linux-androideabi
    steps:
      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
+      - uses: actions-rs/toolchain@v1
        with:
-          toolchain: ${{ env.rust_stable }}
+          toolchain: stable
          target: ${{ matrix.target }}
          override: true
-      - uses: Swatinem/rust-cache@v1
      - uses: actions-rs/cargo@v1
        with:
          use-cross: true
@@ -248,17 +152,16 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_nightly }}
-        uses: actions-rs/toolchain@v1
+      - uses: actions-rs/toolchain@v1
        with:
-          toolchain: ${{ env.rust_nightly }}
-          target: ${{ matrix.target }}
+          toolchain: ${{ env.nightly }}
          override: true
-      - uses: Swatinem/rust-cache@v1
      - name: Install cargo-hack
        run: cargo install cargo-hack
+
      - name: check --each-feature
        run: cargo hack check --all --each-feature -Z avoid-dev-deps
+
      # Try with unstable feature flags
      - name: check --each-feature --unstable
        run: cargo hack check --all --each-feature -Z avoid-dev-deps
@@ -270,12 +173,11 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_min }}
-        uses: actions-rs/toolchain@v1
+      - uses: actions-rs/toolchain@v1
        with:
-          toolchain: ${{ env.rust_min }}
+          toolchain: ${{ env.minrust }}
          override: true
-      - uses: Swatinem/rust-cache@v1
+
      - name: "test --workspace --all-features"
        run: cargo check --workspace --all-features

@@ -284,12 +186,10 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_nightly }}
-        uses: actions-rs/toolchain@v1
+      - uses: actions-rs/toolchain@v1
        with:
-          toolchain: ${{ env.rust_nightly }}
+          toolchain: ${{ env.nightly }}
          override: true
-      - uses: Swatinem/rust-cache@v1
      - name: Install cargo-hack
        run: cargo install cargo-hack
      - name: "check --all-features -Z minimal-versions"
@@ -299,36 +199,24 @@ jobs:
          cargo hack --remove-dev-deps --workspace
          # Update Cargo.lock to minimal version dependencies.
          cargo update -Z minimal-versions
-          cargo hack check --all-features --ignore-private
-      - name: "check --all-features --unstable -Z minimal-versions"
-        env:
-          RUSTFLAGS: --cfg tokio_unstable -Dwarnings
-        run: |
-          # Remove dev-dependencies from Cargo.toml to prevent the next `cargo update`
-          # from determining minimal versions based on dev-dependencies.
-          cargo hack --remove-dev-deps --workspace
-          # Update Cargo.lock to minimal version dependencies.
-          cargo update -Z minimal-versions
-          cargo hack check --all-features --ignore-private
+          cargo check --all-features

  fmt:
    name: fmt
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: ${{ env.rust_stable }}
-          override: true
-          components: rustfmt
-      - uses: Swatinem/rust-cache@v1
+      - name: Install Rust
+        run: rustup update stable
+      - name: Install rustfmt
+        run: rustup component add rustfmt
+
      # Check fmt
      - name: "rustfmt --check"
        # Workaround for rust-lang/cargo#7732
        run: |
-          if ! rustfmt --check --edition 2018 $(git ls-files '*.rs'); then
-            printf "Please run \`rustfmt --edition 2018 \$(git ls-files '*.rs')\` to fix rustfmt errors.\nSee CONTRIBUTING.md for more details.\n" >&2
+          if ! rustfmt --check --edition 2018 $(find . -name '*.rs' -print); then
+            printf "Please run \`rustfmt --edition 2018 \$(find . -name '*.rs' -print)\` to fix rustfmt errors.\nSee CONTRIBUTING.md for more details.\n" >&2
            exit 1
          fi

@@ -337,110 +225,51 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_clippy }}
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: ${{ env.rust_clippy }}
-          override: true
-          components: clippy
-      - uses: Swatinem/rust-cache@v1
+      - name: Install Rust
+        run: rustup update ${{ env.minrust }} && rustup default ${{ env.minrust }}
+      - name: Install clippy
+        run: rustup component add clippy
+
      # Run clippy
      - name: "clippy --all"
-        run: cargo clippy --all --tests --all-features
+        run: cargo clippy --all --tests

  docs:
    name: docs
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_nightly }}
-        uses: actions-rs/toolchain@v1
+      - uses: actions-rs/toolchain@v1
        with:
-          toolchain: ${{ env.rust_nightly }}
+          toolchain: ${{ env.nightly }}
          override: true
-      - uses: Swatinem/rust-cache@v1
-      - name: "doc --lib --all-features"
-        run: cargo doc --lib --no-deps --all-features --document-private-items
-        env:
-          RUSTFLAGS: --cfg docsrs --cfg tokio_unstable
-          RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable -Dwarnings

-  loom-compile:
-    name: build loom tests
+      - name: "doc --lib --all-features"
+        run: cargo doc --lib --no-deps --all-features
+        env:
+          RUSTDOCFLAGS: --cfg docsrs
+
+  loom:
+    name: loom
    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        scope:
+          - --skip loom_pool
+          - loom_pool::group_a
+          - loom_pool::group_b
+          - loom_pool::group_c
+          - loom_pool::group_d
+          - time::driver
    steps:
      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: ${{ env.rust_stable }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
-      - name: build --cfg loom
-        run: cargo test --no-run --lib --features full
+      - name: Install Rust
+        run: rustup update stable
+
+      - name: loom ${{ matrix.scope }}
+        run: cargo test --lib --release --features full -- --nocapture $SCOPE
        working-directory: tokio
        env:
          RUSTFLAGS: --cfg loom --cfg tokio_unstable -Dwarnings
-
-  check-readme:
-    name: Check README
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v2
-      - name: Verify that both READMEs are identical
-        run: diff README.md tokio/README.md
-
-      - name: Verify that Tokio version is up to date in README
-        working-directory: tokio
-        run: grep -q "$(sed '/^version = /!d' Cargo.toml | head -n1)" README.md
-
-  test-hyper:
-    name: Test hyper
-    runs-on: ${{ matrix.os }}
-    strategy:
-      matrix:
-        os:
-          - windows-latest
-          - ubuntu-latest
-          - macos-latest
-    steps:
-      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: ${{ env.rust_stable }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
-      - name: Test hyper
-        run: |
-          set -x
-          git clone https://github.com/hyperium/hyper.git
-          cd hyper
-          # checkout the latest release because HEAD maybe contains breakage.
-          tag=$(git describe --abbrev=0 --tags)
-          git checkout "${tag}"
-          echo '[workspace]' >>Cargo.toml
-          echo '[patch.crates-io]' >>Cargo.toml
-          echo 'tokio = { path = "../tokio" }' >>Cargo.toml
-          echo 'tokio-util = { path = "../tokio-util" }' >>Cargo.toml
-          echo 'tokio-stream = { path = "../tokio-stream" }' >>Cargo.toml
-          echo 'tokio-test = { path = "../tokio-test" }' >>Cargo.toml
-          git diff
-          cargo test --features full
-
-  wasm32-unknown-unknown:
-    name: test tokio for wasm32-unknown-unknown
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: ${{ env.rust_stable }}
-          override: true
-      - uses: Swatinem/rust-cache@v1
-      - name: Install wasm-pack
-        run: curl https://rustwasm.github.io/wasm-pack/installer/init.sh -sSf | sh
-      - name: test tokio
-        run: wasm-pack test --node -- --features "macros sync"
-        working-directory: tokio
+          LOOM_MAX_PREEMPTIONS: 2
+          SCOPE: ${{ matrix.scope }}
@@ -1,14 +0,0 @@
-name: "Pull Request Labeler"
-on:
- pull_request_target
-
-# See .github/labeler.yml file
-
-jobs:
-  triage:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/labeler@v3
-      with:
-        repo-token: "${{ secrets.GITHUB_TOKEN }}"
-        sync-labels: true
@@ -1,45 +0,0 @@
-on:
-  push:
-    branches: ["master", "tokio-*.x"]
-  pull_request:
-    types: [labeled, opened, synchronize, reopened]
-    branches: ["master", "tokio-*.x"]
-
-name: Loom
-
-env:
-  RUSTFLAGS: -Dwarnings
-  RUST_BACKTRACE: 1
-  # Change to specific Rust release to pin
-  rust_stable: stable
-
-jobs:
-  loom:
-    name: loom
-    # base_ref is null when it's not a pull request
-    if: contains(github.event.pull_request.labels.*.name, 'R-loom') || (github.base_ref == null)
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        scope:
-          - --skip loom_pool
-          - loom_pool::group_a
-          - loom_pool::group_b
-          - loom_pool::group_c
-          - loom_pool::group_d
-          - time::driver
-    steps:
-      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
-        with:
-            toolchain: ${{ env.rust_stable }}
-            override: true
-      - uses: Swatinem/rust-cache@v1
-      - name: loom ${{ matrix.scope }}
-        run: cargo test --lib --release --features full -- --nocapture $SCOPE
-        working-directory: tokio
-        env:
-          RUSTFLAGS: --cfg loom --cfg tokio_unstable -Dwarnings
-          LOOM_MAX_PREEMPTIONS: 2
-          SCOPE: ${{ matrix.scope }}
@@ -5,12 +5,6 @@ on:
    branches:
      - master

-env:
-  RUSTFLAGS: -Dwarnings
-  RUST_BACKTRACE: 1
-  # Change to specific Rust release to pin
-  rust_stable: stable
-
 jobs:
  stess-test:
    name: Stress Test
@@ -21,12 +15,9 @@ jobs:
          - simple_echo_tcp
    steps:
      - uses: actions/checkout@v2
-      - name: Install Rust ${{ env.rust_stable }}
-        uses: actions-rs/toolchain@v1
-        with:
-            toolchain: ${{ env.rust_stable }}
-            override: true
-      - uses: Swatinem/rust-cache@v1
+      - name: Install Rust
+        run: rustup update stable
+
      - name: Install Valgrind
        run: |
          sudo apt-get update -y
@@ -1,4 +1,2 @@
 target
 Cargo.lock
-
-.cargo/config.toml
@@ -124,41 +124,23 @@ arguments to many common cargo commands. This section lists some commonly needed
 commands.

 Some commands just need the `--all-features` argument:
-
 ```
 cargo build --all-features
 cargo check --all-features
 cargo test --all-features
 ```
-
 When building documentation normally, the markers that list the features
 required for various parts of Tokio are missing. To build the documentation
 correctly, use this command:
-
 ```
 RUSTDOCFLAGS="--cfg docsrs" cargo +nightly doc --all-features
 ```
-
-To build documentation including Tokio's unstable features, it is necessary to
-pass `--cfg tokio_unstable` to both RustDoc *and* rustc. To build the
-documentation for unstable features, use this command:
-
-```
-RUSTDOCFLAGS="--cfg docsrs --cfg tokio_unstable" RUSTFLAGS="--cfg tokio_unstable" cargo +nightly doc --all-features
-```
-
-There is currently a [bug in cargo] that means documentation cannot be built
-from the root of the workspace. If you `cd` into the `tokio` subdirectory the
-command shown above will work.
-
-[bug in cargo]: https://github.com/rust-lang/cargo/issues/9274
-
 The `cargo fmt` command does not work on the Tokio codebase. You can use the
 command below instead:

 ```
 # Mac or Linux
-rustfmt --check --edition 2018 $(git ls-files '*.rs')
+rustfmt --check --edition 2018 $(find . -name '*.rs' -print)

 # Powershell
 Get-ChildItem . -Filter "*.rs" -Recurse | foreach { rustfmt --check --edition 2018 $_.FullName }
@@ -173,12 +155,6 @@ LOOM_MAX_PREEMPTIONS=1 RUSTFLAGS="--cfg loom" \
    cargo test --lib --release --features full -- --test-threads=1 --nocapture
 ```

-You can run miri tests with
-```
-MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-tag-raw-pointers" PROPTEST_CASES=10 \
-    cargo +nightly miri test --features full --lib
-```
-
 ### Tests

 If the change being proposed alters code (as opposed to only documentation for
@@ -451,14 +427,11 @@ _Adapted from the [Node.js contributing guide][node]_.

 ## Keeping track of issues and PRs

-The Tokio GitHub repository has a lot of issues and PRs to keep track of. This
-section explains the meaning of various labels, as well as our [GitHub
-project][project]. The section is primarily targeted at maintainers.  Most
-contributors aren't able to set these labels.
+The Tokio GitHub repository has a lot of issues and PRs, which is not easy to
+keep track of. This section explains the meaning of various labels, as well as
+our [GitHub project][project]. The section is primarily targeted at maintainers.

-### Area
-
-The area label describes the crates relevant to this issue or PR.
+**Area.** The area label describes the crates relevant to this issue or PR.

 - **A-tokio** This issue concerns the main Tokio crate.
 - **A-tokio-util** This issue concerns the `tokio-util` crate.
@@ -469,7 +442,7 @@ The area label describes the crates relevant to this issue or PR.
   be used for the procedural macros, and not `join!` or `select!`.
 - **A-ci** This issue concerns our GitHub Actions setup.

-### Category
+**Category.** The category label describes the category.

 - **C-bug** This is a bug-report. Bug-fix PRs use `C-enhancement` instead.
 - **C-enhancement** This is a PR that adds a new features.
@@ -487,7 +460,8 @@ The area label describes the crates relevant to this issue or PR.
 - **C-request** A non-feature request, e.g. "please add deprecation notices to
   `-alpha.*` versions of crates"

-### Calls for participation
+**Call for participation.** I don't know why it's called `E-`. Many issues are
+missing a difficulty rating, and you should feel free to add one.

 - **E-help-wanted** Stuff where we want help. Often seen together with `C-bug`
   or `C-feature-accepted`.
@@ -499,13 +473,7 @@ The area label describes the crates relevant to this issue or PR.
 - **E-needs-mvce** This bug is missing a minimal complete and verifiable
   example.

-The "E-" prefix is the same as used in the Rust compiler repository. Some
-issues are missing a difficulty rating, but feel free to ask on our Discord
-server if you want to know how difficult an issue likely is.
-
-### Module
-
-The module label provides a more fine grained categorization than **Area**.
+**Module.** A more fine groaned categorization than area.

 - **M-blocking** Things relevant to `spawn_blocking`, `block_in_place`.
 - **M-codec** The `tokio_util::codec` module.
@@ -518,14 +486,13 @@ The module label provides a more fine grained categorization than **Area**.
 - **M-process** The `tokio::process` module.
 - **M-runtime** The `tokio::runtime` module.
 - **M-signal** The `tokio::signal` module.
+ - **M-stream** The `tokio::stream` module.
 - **M-sync** The `tokio::sync` module.
 - **M-task** The `tokio::task` module.
 - **M-time** The `tokio::time` module.
 - **M-tracing** Tracing support in Tokio.

-### Topic
-
-Some extra information.
+**Topic.** Some extra information.

 - **T-docs** This is about documentation.
 - **T-performance** This is about performance.
@@ -535,34 +502,6 @@ Any label not listed here is not in active use.

 [project]: https://github.com/orgs/tokio-rs/projects/1

-## LTS guarantees
-
-Tokio ≥1.0.0 comes with LTS guarantees:
-
- * A minimum of 5 years of maintenance.
- * A minimum of 3 years before a hypothetical 2.0 release.
-
-The goal of these guarantees is to provide stability to the ecosystem.
-
-## Mininum Supported Rust Version (MSRV)
-
- * All Tokio ≥1.0.0 releases will support at least a 6-month old Rust
-   compiler release.
- * The MSRV will only be increased on 1.x releases.
-
-## Versioning Policy
-
-With Tokio ≥1.0.0:
-
- * Patch (1.\_.x) releases _should only_ contain bug fixes or documentation
-   changes. Besides this, these releases should not substantially change
-   runtime behavior.
- * Minor (1.x) releases may contain new functionality, MSRV increases (see
-   above), minor dependency updates, deprecations, and larger internal
-   implementation changes.
-
-This is as defined by [Semantic Versioning 2.0](https://semver.org/).
-
 ## Releasing

 Since the Tokio project consists of a number of crates, many of which depend on
@@ -595,8 +534,9 @@ When releasing a new version of a crate, follow these steps:
 2. **Update Cargo metadata.** After releasing any path dependencies, update the
   `version` field in `Cargo.toml` to the new version, and the `documentation`
   field to the docs.rs URL of the new version.
-3. **Update other documentation links.** Update the "Documentation" link in the
-   crate's `README.md` to point to the docs.rs URL of the new version.
+3. **Update other documentation links.** Update the `#![doc(html_root_url)]`
+   attribute in the crate's `lib.rs` and the "Documentation" link in the crate's
+   `README.md` to point to the docs.rs URL of the new version.
 4. **Update the changelog for the crate.** Each crate in the Tokio repository
   has its own `CHANGELOG.md` in that crate's subdirectory. Any changes to that
   crate since the last release should be added to the changelog. Change
@@ -4,7 +4,6 @@ members = [
  "tokio",
  "tokio-macros",
  "tokio-test",
-  "tokio-stream",
  "tokio-util",

  # Internal
@@ -1,4 +1,4 @@
-Copyright (c) 2022 Tokio Contributors
+Copyright (c) 2019 Tokio Contributors

 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated
@@ -29,6 +29,7 @@ the Rust programming language. It is:
 [Website](https://tokio.rs) |
 [Guides](https://tokio.rs/tokio/tutorial) |
 [API Docs](https://docs.rs/tokio/latest/tokio) |
+[Roadmap](https://github.com/tokio-rs/tokio/blob/master/ROADMAP.md) |
 [Chat](https://discord.gg/tokio)

 ## Overview
@@ -50,23 +51,15 @@ an asynchronous application.

 ## Example

-A basic TCP echo server with Tokio.
-
-Make sure you activated the full features of the tokio crate on Cargo.toml:
-
-```toml
-[dependencies]
-tokio = { version = "1.17.0", features = ["full"] }
-```
-Then, on your main.rs:
+A basic TCP echo server with Tokio:

 ```rust,no_run
 use tokio::net::TcpListener;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::prelude::*;

 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let listener = TcpListener::bind("127.0.0.1:8080").await?;
+    let mut listener = TcpListener::bind("127.0.0.1:8080").await?;

    loop {
        let (mut socket, _) = listener.accept().await?;
@@ -140,7 +133,8 @@ several other libraries, including:

 * [`tower`]: A library of modular and reusable components for building robust networking clients and servers.

-* [`tracing`] (formerly `tokio-trace`): A framework for application-level tracing and async-aware diagnostics.
+* [`tracing`] (formerly `tokio-trace`): A framework for application-level
+  tracing and async-aware diagnostics.

 * [`rdbc`]: A Rust database connectivity library for MySQL, Postgres and SQLite.

@@ -163,36 +157,9 @@ several other libraries, including:

 ## Supported Rust Versions

-Tokio will keep a rolling MSRV (minimum supported rust version) policy of **at
-least** 6 months. When increasing the MSRV, the new Rust version must have been
-released at least six months ago. The current MSRV is 1.49.0.
-
-## Release schedule
-
-Tokio doesn't follow a fixed release schedule, but we typically make one to two
-new minor releases each month. We make patch releases for bugfixes as necessary.
-
-## Bug patching policy
-
-For the purposes of making patch releases with bugfixes, we have designated
-certain minor releases as LTS (long term support) releases. Whenever a bug
-warrants a patch release with a fix for the bug, it will be backported and
-released as a new patch release for each LTS minor version. Our current LTS
-releases are:
-
- * `1.8.x` - LTS release until February 2022.
- * `1.14.x` - LTS release until June 2022.
-
-Each LTS release will continue to receive backported fixes for at least half a
-year. If you wish to use a fixed minor release in your project, we recommend
-that you use an LTS release.
-
-To use a fixed minor version, you can specify the version with a tilde. For
-example, to specify that you wish to use the newest `1.8.x` patch release, you
-can use the following dependency specification:
-```text
-tokio = { version = "~1.8", features = [...] }
-```
+Tokio is built against the latest stable release. The minimum supported version is 1.45.
+The current Tokio version is not guaranteed to build on Rust versions earlier than the
+minimum supported version.

 ## License

@@ -0,0 +1,67 @@
+# Tokio Roadmap
+
+## A Roadmap to 1.0
+
+The question of "why not 1.0?" has come up a few times. After all, Tokio 0.1 has
+been stable for three years. The short answer: because it isn't time. There is
+nobody who would rather ship a Tokio 1.0 than us. It also isn't something to rush.
+
+After all, `async / await` only landed in the stable Rust channel weeks ago.
+There has been no significant production validation yet, except maybe fuchsia
+and that seems like a fairly specialized use case. This release of Tokio
+includes significant new code and new strategies with feature flags. Also, there
+are still big open questions, such as the [proposed changes][pr-1744] to
+`AsyncRead` and `AsyncWrite`.
+
+Tokio 1.0 will be released as soon as the APIs are proven to handle real-world
+production cases.
+
+### Tokio 1.0 in Q3 2020 with LTS support
+
+The Tokio 1.0 release will be **no later** than Q3 2020. It will also come with
+"long-term support" guarantees:
+
+* A minimum of 5 years of maintenance.
+* A minimum of 3 years before a hypothetical 2.0 release.
+
+When Tokio 1.0 is released in Q3 2020, on-going support, security fixes, and
+critical bug fixes are guaranteed until **at least** Q3 2025. Tokio 2.0 will not
+be released until **at least** Q3 2023 (though, ideally there will never be a
+Tokio 2.0 release).
+
+### How to get there
+
+While Tokio 0.1 probably should have been a 1.0, Tokio 0.2 will be a **true**
+0.2 release. There will be breaking change releases every 2 ~ 3 months until 1.0.
+These changes will be **much** smaller than going from 0.1 -> 0.2. It is
+expected that the 1.0 release will look a lot like 0.2.
+
+### What is expected to change
+
+The biggest change will be the `AsyncRead` and `AsyncWrite` traits. Based on
+experience gained over the past 3 years, there are a couple of issues to
+address:
+
+* Be able to **safely** use uninitialized memory as a read buffer.
+* Practical read vectored and write vectored APIs.
+
+There are a few strategies to solve these problems. These strategies need to be
+investigated and the solution validated. You can see [this comment][pr-1744-comment] for a
+detailed statement of the problem.
+
+The other major change, which has been in the works for a while, is updating
+Mio. Mio 0.6 was first released almost 4 years ago and has not had a breaking
+change since. Mio 0.7 has been in the works for a while. It includes a full
+rewrite of the windows support as well as a refined API. More will be written
+about this shortly.
+
+Finally, now that the API is starting to stabilize, effort will be put into
+documentation. Tokio 0.2 is being released before updating the website and many
+of the old content will no longer be relevant. In the coming weeks, expect to
+see updates there.
+
+So, we have our work cut out for us. We hope you enjoy this 0.2 release and are
+looking forward to your feedback and help.
+
+[pr-1744]: https://github.com/tokio-rs/tokio/pull/1744
+[pr-1744-comment]: https://github.com/tokio-rs/tokio/pull/1744#issuecomment-553575438
@@ -5,13 +5,9 @@ publish = false
 edition = "2018"

 [dependencies]
-tokio = { version = "1.5.0", path = "../tokio", features = ["full"] }
+tokio = { version = "0.3.0", path = "../tokio", features = ["full"] }
 bencher = "0.1.5"

-[dev-dependencies]
-tokio-util = { version = "0.7.0", path = "../tokio-util", features = ["full"] }
-tokio-stream = { path = "../tokio-stream" }
-
 [target.'cfg(unix)'.dependencies]
 libc = "0.2.42"

@@ -45,8 +41,3 @@ harness = false
 name = "signal"
 path = "signal.rs"
 harness = false
-
-[[bench]]
-name = "fs"
-path = "fs.rs"
-harness = false
@@ -1,103 +0,0 @@
-#![cfg(unix)]
-
-use tokio_stream::StreamExt;
-
-use tokio::fs::File;
-use tokio::io::AsyncReadExt;
-use tokio_util::codec::{BytesCodec, FramedRead /*FramedWrite*/};
-
-use bencher::{benchmark_group, benchmark_main, Bencher};
-
-use std::fs::File as StdFile;
-use std::io::Read as StdRead;
-
-fn rt() -> tokio::runtime::Runtime {
-    tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(2)
-        .build()
-        .unwrap()
-}
-
-const BLOCK_COUNT: usize = 1_000;
-
-const BUFFER_SIZE: usize = 4096;
-const DEV_ZERO: &str = "/dev/zero";
-
-fn async_read_codec(b: &mut Bencher) {
-    let rt = rt();
-
-    b.iter(|| {
-        let task = || async {
-            let file = File::open(DEV_ZERO).await.unwrap();
-            let mut input_stream = FramedRead::with_capacity(file, BytesCodec::new(), BUFFER_SIZE);
-
-            for _i in 0..BLOCK_COUNT {
-                let _bytes = input_stream.next().await.unwrap();
-            }
-        };
-
-        rt.block_on(task());
-    });
-}
-
-fn async_read_buf(b: &mut Bencher) {
-    let rt = rt();
-
-    b.iter(|| {
-        let task = || async {
-            let mut file = File::open(DEV_ZERO).await.unwrap();
-            let mut buffer = [0u8; BUFFER_SIZE];
-
-            for _i in 0..BLOCK_COUNT {
-                let count = file.read(&mut buffer).await.unwrap();
-                if count == 0 {
-                    break;
-                }
-            }
-        };
-
-        rt.block_on(task());
-    });
-}
-
-fn async_read_std_file(b: &mut Bencher) {
-    let rt = rt();
-
-    let task = || async {
-        let mut file = tokio::task::block_in_place(|| Box::pin(StdFile::open(DEV_ZERO).unwrap()));
-
-        for _i in 0..BLOCK_COUNT {
-            let mut buffer = [0u8; BUFFER_SIZE];
-            let mut file_ref = file.as_mut();
-
-            tokio::task::block_in_place(move || {
-                file_ref.read_exact(&mut buffer).unwrap();
-            });
-        }
-    };
-
-    b.iter(|| {
-        rt.block_on(task());
-    });
-}
-
-fn sync_read(b: &mut Bencher) {
-    b.iter(|| {
-        let mut file = StdFile::open(DEV_ZERO).unwrap();
-        let mut buffer = [0u8; BUFFER_SIZE];
-
-        for _i in 0..BLOCK_COUNT {
-            file.read_exact(&mut buffer).unwrap();
-        }
-    });
-}
-
-benchmark_group!(
-    file,
-    async_read_std_file,
-    async_read_buf,
-    async_read_codec,
-    sync_read
-);
-
-benchmark_main!(file);
@@ -1,4 +1,4 @@
-//! Benchmark implementation details of the threaded scheduler. These benches are
+//! Benchmark implementation details of the theaded scheduler. These benches are
 //! intended to be used as a form of regression testing and not as a general
 //! purpose benchmark demonstrating real-world performance.

@@ -2,83 +2,63 @@
 //! This essentially measure the time to enqueue a task in the local and remote
 //! case.

-#[macro_use]
-extern crate bencher;
-
 use bencher::{black_box, Bencher};

 async fn work() -> usize {
    let val = 1 + 1;
-    tokio::task::yield_now().await;
    black_box(val)
 }

-fn basic_scheduler_spawn(bench: &mut Bencher) {
+fn basic_scheduler_local_spawn(bench: &mut Bencher) {
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
-    bench.iter(|| {
-        runtime.block_on(async {
+    runtime.block_on(async {
+        bench.iter(|| {
            let h = tokio::spawn(work());
-            assert_eq!(h.await.unwrap(), 2);
-        });
+            black_box(h);
+        })
    });
 }

-fn basic_scheduler_spawn10(bench: &mut Bencher) {
+fn threaded_scheduler_local_spawn(bench: &mut Bencher) {
    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
-    bench.iter(|| {
-        runtime.block_on(async {
-            let mut handles = Vec::with_capacity(10);
-            for _ in 0..10 {
-                handles.push(tokio::spawn(work()));
-            }
-            for handle in handles {
-                assert_eq!(handle.await.unwrap(), 2);
-            }
-        });
-    });
-}
-
-fn threaded_scheduler_spawn(bench: &mut Bencher) {
-    let runtime = tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(1)
-        .build()
-        .unwrap();
-    bench.iter(|| {
-        runtime.block_on(async {
+    runtime.block_on(async {
+        bench.iter(|| {
            let h = tokio::spawn(work());
-            assert_eq!(h.await.unwrap(), 2);
-        });
+            black_box(h);
+        })
    });
 }

-fn threaded_scheduler_spawn10(bench: &mut Bencher) {
-    let runtime = tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(1)
+fn basic_scheduler_remote_spawn(bench: &mut Bencher) {
+    let runtime = tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap();
+
    bench.iter(|| {
-        runtime.block_on(async {
-            let mut handles = Vec::with_capacity(10);
-            for _ in 0..10 {
-                handles.push(tokio::spawn(work()));
-            }
-            for handle in handles {
-                assert_eq!(handle.await.unwrap(), 2);
-            }
-        });
+        let h = runtime.spawn(work());
+        black_box(h);
+    });
+}
+
+fn threaded_scheduler_remote_spawn(bench: &mut Bencher) {
+    let runtime = tokio::runtime::Builder::new_multi_thread().build().unwrap();
+
+    bench.iter(|| {
+        let h = runtime.spawn(work());
+        black_box(h);
    });
 }

 bencher::benchmark_group!(
    spawn,
-    basic_scheduler_spawn,
-    basic_scheduler_spawn10,
-    threaded_scheduler_spawn,
-    threaded_scheduler_spawn10,
+    basic_scheduler_local_spawn,
+    threaded_scheduler_local_spawn,
+    basic_scheduler_remote_spawn,
+    threaded_scheduler_remote_spawn
 );

 bencher::benchmark_main!(spawn);
@@ -30,26 +30,22 @@ fn create_100_000_medium(b: &mut Bencher) {
 }

 fn send_medium(b: &mut Bencher) {
-    let rt = rt();
-
    b.iter(|| {
        let (tx, mut rx) = mpsc::channel::<Medium>(1000);

-        let _ = rt.block_on(tx.send([0; 64]));
+        let _ = tx.try_send([0; 64]);

-        rt.block_on(rx.recv()).unwrap();
+        rx.try_recv().unwrap();
    });
 }

 fn send_large(b: &mut Bencher) {
-    let rt = rt();
-
    b.iter(|| {
        let (tx, mut rx) = mpsc::channel::<Large>(1000);

-        let _ = rt.block_on(tx.send([[0; 64]; 64]));
+        let _ = tx.try_send([[0; 64]; 64]);

-        rt.block_on(rx.recv()).unwrap();
+        rx.try_recv().unwrap();
    });
 }

@@ -5,27 +5,20 @@ publish = false
 edition = "2018"

 # If you copy one of the examples into a new project, you should be using
-# [dependencies] instead, and delete the **path**.
+# [dependencies] instead.
 [dev-dependencies]
-tokio = { version = "1.0.0", path = "../tokio", features = ["full", "tracing"] }
-tokio-util = { version = "0.7.0", path = "../tokio-util", features = ["full"] }
-tokio-stream = { version = "0.1", path = "../tokio-stream" }
-
+tokio = { version = "0.3.0", path = "../tokio", features = ["full", "tracing"] }
 tracing = "0.1"
-tracing-subscriber = { version = "0.3.1", default-features = false, features = ["fmt", "ansi", "env-filter", "tracing-log"] }
-bytes = "1.0.0"
-futures = { version = "0.3.0", features = ["thread-pool"]}
+tracing-subscriber = { version = "0.2.7", default-features = false, features = ["fmt", "ansi", "env-filter", "chrono", "tracing-log"] }
+tokio-util = { version = "0.5.0", path = "../tokio-util", features = ["full"] }
+bytes = "0.6"
+futures = "0.3.0"
 http = "0.2"
 serde = "1.0"
 serde_derive = "1.0"
 serde_json = "1.0"
 httparse = "1.0"
-httpdate = "1.0"
-once_cell = "1.5.2"
-rand = "0.8.3"
-
-[target.'cfg(windows)'.dev-dependencies.winapi]
-version = "0.3.8"
+time = "0.1"

 [[example]]
 name = "chat"
@@ -70,24 +63,3 @@ path = "udp-codec.rs"
 [[example]]
 name = "tinyhttp"
 path = "tinyhttp.rs"
-
-[[example]]
-name = "custom-executor"
-path = "custom-executor.rs"
-
-
-[[example]]
-name = "custom-executor-tokio-context"
-path = "custom-executor-tokio-context.rs"
-
-[[example]]
-name = "named-pipe"
-path = "named-pipe.rs"
-
-[[example]]
-name = "named-pipe-ready"
-path = "named-pipe-ready.rs"
-
-[[example]]
-name = "named-pipe-multi-client"
-path = "named-pipe-multi-client.rs"
@@ -27,9 +27,9 @@
 #![warn(rust_2018_idioms)]

 use tokio::net::{TcpListener, TcpStream};
+use tokio::stream::{Stream, StreamExt};
 use tokio::sync::{mpsc, Mutex};
-use tokio_stream::StreamExt;
-use tokio_util::codec::{Framed, LinesCodec};
+use tokio_util::codec::{Framed, LinesCodec, LinesCodecError};

 use futures::SinkExt;
 use std::collections::HashMap;
@@ -37,7 +37,9 @@ use std::env;
 use std::error::Error;
 use std::io;
 use std::net::SocketAddr;
+use std::pin::Pin;
 use std::sync::Arc;
+use std::task::{Context, Poll};

 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
@@ -166,6 +168,43 @@ impl Peer {
    }
 }

+#[derive(Debug)]
+enum Message {
+    /// A message that should be broadcasted to others.
+    Broadcast(String),
+
+    /// A message that should be received by a client
+    Received(String),
+}
+
+// Peer implements `Stream` in a way that polls both the `Rx`, and `Framed` types.
+// A message is produced whenever an event is ready until the `Framed` stream returns `None`.
+impl Stream for Peer {
+    type Item = Result<Message, LinesCodecError>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        // First poll the `UnboundedReceiver`.
+
+        if let Poll::Ready(Some(v)) = Pin::new(&mut self.rx).poll_next(cx) {
+            return Poll::Ready(Some(Ok(Message::Received(v))));
+        }
+
+        // Secondly poll the `Framed` stream.
+        let result: Option<_> = futures::ready!(Pin::new(&mut self.lines).poll_next(cx));
+
+        Poll::Ready(match result {
+            // We've received a message we should broadcast to others.
+            Some(Ok(message)) => Some(Ok(Message::Broadcast(message))),
+
+            // An error occurred.
+            Some(Err(e)) => Some(Err(e)),
+
+            // The stream has been exhausted.
+            None => None,
+        })
+    }
+}
+
 /// Process an individual chat client
 async fn process(
    state: Arc<Mutex<Shared>>,
@@ -199,32 +238,28 @@ async fn process(
    }

    // Process incoming messages until our stream is exhausted by a disconnect.
-    loop {
-        tokio::select! {
-            // A message was received from a peer. Send it to the current user.
-            Some(msg) = peer.rx.recv() => {
+    while let Some(result) = peer.next().await {
+        match result {
+            // A message was received from the current user, we should
+            // broadcast this message to the other users.
+            Ok(Message::Broadcast(msg)) => {
+                let mut state = state.lock().await;
+                let msg = format!("{}: {}", username, msg);
+
+                state.broadcast(addr, &msg).await;
+            }
+            // A message was received from a peer. Send it to the
+            // current user.
+            Ok(Message::Received(msg)) => {
                peer.lines.send(&msg).await?;
            }
-            result = peer.lines.next() => match result {
-                // A message was received from the current user, we should
-                // broadcast this message to the other users.
-                Some(Ok(msg)) => {
-                    let mut state = state.lock().await;
-                    let msg = format!("{}: {}", username, msg);
-
-                    state.broadcast(addr, &msg).await;
-                }
-                // An error occurred.
-                Some(Err(e)) => {
-                    tracing::error!(
-                        "an error occurred while processing messages for {}; error = {:?}",
-                        username,
-                        e
-                    );
-                }
-                // The stream has been exhausted.
-                None => break,
-            },
+            Err(e) => {
+                tracing::error!(
+                    "an error occurred while processing messages for {}; error = {:?}",
+                    username,
+                    e
+                );
+            }
        }
    }

@@ -92,7 +92,7 @@ mod tcp {

 mod udp {
    use bytes::Bytes;
-    use futures::{Sink, SinkExt, Stream, StreamExt};
+    use futures::{future, Sink, SinkExt, Stream, StreamExt};
    use std::error::Error;
    use std::io;
    use std::net::SocketAddr;
@@ -114,7 +114,7 @@ mod udp {
        let socket = UdpSocket::bind(&bind_addr).await?;
        socket.connect(addr).await?;

-        tokio::try_join!(send(stdin, &socket), recv(stdout, &socket))?;
+        future::try_join(send(stdin, &socket), recv(stdout, &socket)).await?;

        Ok(())
    }
@@ -1,32 +0,0 @@
-// This example shows how to use the tokio runtime with any other executor
-//
-//It takes advantage from RuntimeExt which provides the extension to customize your
-//runtime.
-
-use tokio::net::TcpListener;
-use tokio::runtime::Builder;
-use tokio::sync::oneshot;
-use tokio_util::context::RuntimeExt;
-
-fn main() {
-    let (tx, rx) = oneshot::channel();
-    let rt1 = Builder::new_multi_thread()
-        .worker_threads(1)
-        // no timer!
-        .build()
-        .unwrap();
-    let rt2 = Builder::new_multi_thread()
-        .worker_threads(1)
-        .enable_all()
-        .build()
-        .unwrap();
-
-    // Without the `HandleExt.wrap()` there would be a panic because there is
-    // no timer running, since it would be referencing runtime r1.
-    let _ = rt1.block_on(rt2.wrap(async move {
-        let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
-        println!("addr: {:?}", listener.local_addr());
-        tx.send(()).unwrap();
-    }));
-    futures::executor::block_on(rx).unwrap();
-}
@@ -1,56 +0,0 @@
-// This example shows how to use the tokio runtime with any other executor
-//
-// The main components are a spawn fn that will wrap futures in a special future
-// that will always enter the tokio context on poll. This only spawns one extra thread
-// to manage and run the tokio drivers in the background.
-
-use tokio::net::TcpListener;
-use tokio::sync::oneshot;
-
-fn main() {
-    let (tx, rx) = oneshot::channel();
-
-    my_custom_runtime::spawn(async move {
-        let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
-
-        println!("addr: {:?}", listener.local_addr());
-
-        tx.send(()).unwrap();
-    });
-
-    futures::executor::block_on(rx).unwrap();
-}
-
-mod my_custom_runtime {
-    use once_cell::sync::Lazy;
-    use std::future::Future;
-    use tokio_util::context::TokioContext;
-
-    pub fn spawn(f: impl Future<Output = ()> + Send + 'static) {
-        EXECUTOR.spawn(f);
-    }
-
-    struct ThreadPool {
-        inner: futures::executor::ThreadPool,
-        rt: tokio::runtime::Runtime,
-    }
-
-    static EXECUTOR: Lazy<ThreadPool> = Lazy::new(|| {
-        // Spawn tokio runtime on a single background thread
-        // enabling IO and timers.
-        let rt = tokio::runtime::Builder::new_multi_thread()
-            .enable_all()
-            .build()
-            .unwrap();
-        let inner = futures::executor::ThreadPool::builder().create().unwrap();
-
-        ThreadPool { inner, rt }
-    });
-
-    impl ThreadPool {
-        fn spawn(&self, f: impl Future<Output = ()> + Send + 'static) {
-            let handle = self.rt.handle().clone();
-            self.inner.spawn_ok(TokioContext::new(f, handle));
-        }
-    }
-}
@@ -55,7 +55,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
        // which will allow all of our clients to be processed concurrently.

        tokio::spawn(async move {
-            let mut buf = vec![0; 1024];
+            let mut buf = [0; 1024];

            // In a loop, read data from the socket and write the data back.
            loop {
@@ -1,98 +0,0 @@
-use std::io;
-
-#[cfg(windows)]
-async fn windows_main() -> io::Result<()> {
-    use std::time::Duration;
-    use tokio::io::{AsyncReadExt, AsyncWriteExt};
-    use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions};
-    use tokio::time;
-    use winapi::shared::winerror;
-
-    const PIPE_NAME: &str = r"\\.\pipe\named-pipe-multi-client";
-    const N: usize = 10;
-
-    // The first server needs to be constructed early so that clients can
-    // be correctly connected. Otherwise a waiting client will error.
-    //
-    // Here we also make use of `first_pipe_instance`, which will ensure
-    // that there are no other servers up and running already.
-    let mut server = ServerOptions::new()
-        .first_pipe_instance(true)
-        .create(PIPE_NAME)?;
-
-    let server = tokio::spawn(async move {
-        // Artificial workload.
-        time::sleep(Duration::from_secs(1)).await;
-
-        for _ in 0..N {
-            // Wait for client to connect.
-            server.connect().await?;
-            let mut inner = server;
-
-            // Construct the next server to be connected before sending the one
-            // we already have of onto a task. This ensures that the server
-            // isn't closed (after it's done in the task) before a new one is
-            // available. Otherwise the client might error with
-            // `io::ErrorKind::NotFound`.
-            server = ServerOptions::new().create(PIPE_NAME)?;
-
-            let _ = tokio::spawn(async move {
-                let mut buf = vec![0u8; 4];
-                inner.read_exact(&mut buf).await?;
-                inner.write_all(b"pong").await?;
-                Ok::<_, io::Error>(())
-            });
-        }
-
-        Ok::<_, io::Error>(())
-    });
-
-    let mut clients = Vec::new();
-
-    for _ in 0..N {
-        clients.push(tokio::spawn(async move {
-            // This showcases a generic connect loop.
-            //
-            // We immediately try to create a client, if it's not found or
-            // the pipe is busy we use the specialized wait function on the
-            // client builder.
-            let mut client = loop {
-                match ClientOptions::new().open(PIPE_NAME) {
-                    Ok(client) => break client,
-                    Err(e) if e.raw_os_error() == Some(winerror::ERROR_PIPE_BUSY as i32) => (),
-                    Err(e) => return Err(e),
-                }
-
-                time::sleep(Duration::from_millis(5)).await;
-            };
-
-            let mut buf = [0u8; 4];
-            client.write_all(b"ping").await?;
-            client.read_exact(&mut buf).await?;
-            Ok::<_, io::Error>(buf)
-        }));
-    }
-
-    for client in clients {
-        let result = client.await?;
-        assert_eq!(&result?[..], b"pong");
-    }
-
-    server.await??;
-    Ok(())
-}
-
-#[tokio::main]
-async fn main() -> io::Result<()> {
-    #[cfg(windows)]
-    {
-        windows_main().await?;
-    }
-
-    #[cfg(not(windows))]
-    {
-        println!("Named pipes are only supported on Windows!");
-    }
-
-    Ok(())
-}
@@ -1,161 +0,0 @@
-use std::io;
-
-#[cfg(windows)]
-async fn windows_main() -> io::Result<()> {
-    use tokio::io::Interest;
-    use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions};
-
-    const PIPE_NAME: &str = r"\\.\pipe\named-pipe-single-client";
-
-    let server = ServerOptions::new().create(PIPE_NAME)?;
-
-    let server = tokio::spawn(async move {
-        // Note: we wait for a client to connect.
-        server.connect().await?;
-
-        let buf = {
-            let mut read_buf = [0u8; 5];
-            let mut read_buf_cursor = 0;
-
-            loop {
-                server.readable().await?;
-
-                let buf = &mut read_buf[read_buf_cursor..];
-
-                match server.try_read(buf) {
-                    Ok(n) => {
-                        read_buf_cursor += n;
-
-                        if read_buf_cursor == read_buf.len() {
-                            break;
-                        }
-                    }
-                    Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
-                        continue;
-                    }
-                    Err(e) => {
-                        return Err(e);
-                    }
-                }
-            }
-
-            read_buf
-        };
-
-        {
-            let write_buf = b"pong\n";
-            let mut write_buf_cursor = 0;
-
-            loop {
-                let buf = &write_buf[write_buf_cursor..];
-
-                if buf.is_empty() {
-                    break;
-                }
-
-                server.writable().await?;
-
-                match server.try_write(buf) {
-                    Ok(n) => {
-                        write_buf_cursor += n;
-                    }
-                    Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
-                        continue;
-                    }
-                    Err(e) => {
-                        return Err(e);
-                    }
-                }
-            }
-        }
-
-        Ok::<_, io::Error>(buf)
-    });
-
-    let client = tokio::spawn(async move {
-        // There's no need to use a connect loop here, since we know that the
-        // server is already up - `open` was called before spawning any of the
-        // tasks.
-        let client = ClientOptions::new().open(PIPE_NAME)?;
-
-        let mut read_buf = [0u8; 5];
-        let mut read_buf_cursor = 0;
-        let write_buf = b"ping\n";
-        let mut write_buf_cursor = 0;
-
-        loop {
-            let mut interest = Interest::READABLE;
-            if write_buf_cursor < write_buf.len() {
-                interest |= Interest::WRITABLE;
-            }
-
-            let ready = client.ready(interest).await?;
-
-            if ready.is_readable() {
-                let buf = &mut read_buf[read_buf_cursor..];
-
-                match client.try_read(buf) {
-                    Ok(n) => {
-                        read_buf_cursor += n;
-
-                        if read_buf_cursor == read_buf.len() {
-                            break;
-                        }
-                    }
-                    Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
-                        continue;
-                    }
-                    Err(e) => {
-                        return Err(e);
-                    }
-                }
-            }
-
-            if ready.is_writable() {
-                let buf = &write_buf[write_buf_cursor..];
-
-                if buf.is_empty() {
-                    continue;
-                }
-
-                match client.try_write(buf) {
-                    Ok(n) => {
-                        write_buf_cursor += n;
-                    }
-                    Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
-                        continue;
-                    }
-                    Err(e) => {
-                        return Err(e);
-                    }
-                }
-            }
-        }
-
-        let buf = String::from_utf8_lossy(&read_buf).into_owned();
-
-        Ok::<_, io::Error>(buf)
-    });
-
-    let (server, client) = tokio::try_join!(server, client)?;
-
-    assert_eq!(server?, *b"ping\n");
-    assert_eq!(client?, "pong\n");
-
-    Ok(())
-}
-
-#[tokio::main]
-async fn main() -> io::Result<()> {
-    #[cfg(windows)]
-    {
-        windows_main().await?;
-    }
-
-    #[cfg(not(windows))]
-    {
-        println!("Named pipes are only supported on Windows!");
-    }
-
-    Ok(())
-}
@@ -1,60 +0,0 @@
-use std::io;
-
-#[cfg(windows)]
-async fn windows_main() -> io::Result<()> {
-    use tokio::io::AsyncWriteExt;
-    use tokio::io::{AsyncBufReadExt, BufReader};
-    use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions};
-
-    const PIPE_NAME: &str = r"\\.\pipe\named-pipe-single-client";
-
-    let server = ServerOptions::new().create(PIPE_NAME)?;
-
-    let server = tokio::spawn(async move {
-        // Note: we wait for a client to connect.
-        server.connect().await?;
-
-        let mut server = BufReader::new(server);
-
-        let mut buf = String::new();
-        server.read_line(&mut buf).await?;
-        server.write_all(b"pong\n").await?;
-        Ok::<_, io::Error>(buf)
-    });
-
-    let client = tokio::spawn(async move {
-        // There's no need to use a connect loop here, since we know that the
-        // server is already up - `open` was called before spawning any of the
-        // tasks.
-        let client = ClientOptions::new().open(PIPE_NAME)?;
-
-        let mut client = BufReader::new(client);
-
-        let mut buf = String::new();
-        client.write_all(b"ping\n").await?;
-        client.read_line(&mut buf).await?;
-        Ok::<_, io::Error>(buf)
-    });
-
-    let (server, client) = tokio::try_join!(server, client)?;
-
-    assert_eq!(server?, "ping\n");
-    assert_eq!(client?, "pong\n");
-
-    Ok(())
-}
-
-#[tokio::main]
-async fn main() -> io::Result<()> {
-    #[cfg(windows)]
-    {
-        windows_main().await?;
-    }
-
-    #[cfg(not(windows))]
-    {
-        println!("Named pipes are only supported on Windows!");
-    }
-
-    Ok(())
-}
@@ -55,7 +55,7 @@
 #![warn(rust_2018_idioms)]

 use tokio::net::TcpListener;
-use tokio_stream::StreamExt;
+use tokio::stream::StreamExt;
 use tokio_util::codec::{BytesCodec, Decoder};

 use std::env;
@@ -26,6 +26,7 @@ use tokio::io;
 use tokio::io::AsyncWriteExt;
 use tokio::net::{TcpListener, TcpStream};

+use futures::future::try_join;
 use futures::FutureExt;
 use std::env;
 use std::error::Error;
@@ -73,7 +74,7 @@ async fn transfer(mut inbound: TcpStream, proxy_addr: String) -> Result<(), Box<
        wi.shutdown().await
    };

-    tokio::try_join!(client_to_server, server_to_client)?;
+    try_join(client_to_server, server_to_client).await?;

    Ok(())
 }
@@ -42,7 +42,7 @@
 #![warn(rust_2018_idioms)]

 use tokio::net::TcpListener;
-use tokio_stream::StreamExt;
+use tokio::stream::StreamExt;
 use tokio_util::codec::{Framed, LinesCodec};

 use futures::SinkExt;
@@ -149,7 +149,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 }

 fn handle_request(line: &str, db: &Arc<Database>) -> Response {
-    let request = match Request::parse(line) {
+    let request = match Request::parse(&line) {
        Ok(req) => req,
        Err(e) => return Response::Error { msg: e },
    };
@@ -20,7 +20,7 @@ use http::{header::HeaderValue, Request, Response, StatusCode};
 extern crate serde_derive;
 use std::{env, error::Error, fmt, io};
 use tokio::net::{TcpListener, TcpStream};
-use tokio_stream::StreamExt;
+use tokio::stream::StreamExt;
 use tokio_util::codec::{Decoder, Encoder, Framed};

 #[tokio::main]
@@ -221,9 +221,8 @@ mod date {
    use std::cell::RefCell;
    use std::fmt::{self, Write};
    use std::str;
-    use std::time::SystemTime;

-    use httpdate::HttpDate;
+    use time::{self, Duration};

    pub struct Now(());

@@ -253,26 +252,22 @@ mod date {
    struct LastRenderedNow {
        bytes: [u8; 128],
        amt: usize,
-        unix_date: u64,
+        next_update: time::Timespec,
    }

    thread_local!(static LAST: RefCell<LastRenderedNow> = RefCell::new(LastRenderedNow {
        bytes: [0; 128],
        amt: 0,
-        unix_date: 0,
+        next_update: time::Timespec::new(0, 0),
    }));

    impl fmt::Display for Now {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            LAST.with(|cache| {
                let mut cache = cache.borrow_mut();
-                let now = SystemTime::now();
-                let now_unix = now
-                    .duration_since(SystemTime::UNIX_EPOCH)
-                    .map(|since_epoch| since_epoch.as_secs())
-                    .unwrap_or(0);
-                if cache.unix_date != now_unix {
-                    cache.update(now, now_unix);
+                let now = time::get_time();
+                if now >= cache.next_update {
+                    cache.update(now);
                }
                f.write_str(cache.buffer())
            })
@@ -284,10 +279,11 @@ mod date {
            str::from_utf8(&self.bytes[..self.amt]).unwrap()
        }

-        fn update(&mut self, now: SystemTime, now_unix: u64) {
+        fn update(&mut self, now: time::Timespec) {
            self.amt = 0;
-            self.unix_date = now_unix;
-            write!(LocalBuffer(self), "{}", HttpDate::from(now)).unwrap();
+            write!(LocalBuffer(self), "{}", time::at(now).rfc822()).unwrap();
+            self.next_update = now + Duration::seconds(1);
+            self.next_update.nsec = 0;
        }
    }

@@ -9,8 +9,8 @@
 #![warn(rust_2018_idioms)]

 use tokio::net::UdpSocket;
+use tokio::stream::StreamExt;
 use tokio::{io, time};
-use tokio_stream::StreamExt;
 use tokio_util::codec::BytesCodec;
 use tokio_util::udp::UdpFramed;

@@ -45,7 +45,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
    let b = pong(&mut b);

    // Run both futures simultaneously of `a` and `b` sending messages back and forth.
-    match tokio::try_join!(a, b) {
+    match futures::future::try_join(a, b).await {
        Err(e) => println!("an error occurred; error = {:?}", e),
        _ => println!("done!"),
    }
@@ -8,7 +8,7 @@ publish = false
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

 [dependencies]
-tokio = { path = "../tokio/", features = ["full"] }
+tokio = {path = "../tokio/", features = ["full"]}

 [dev-dependencies]
-rand = "0.8"
+rand = "0.7.3"
@@ -4,4 +4,4 @@ error: The default runtime flavor is `multi_thread`, but the `rt-multi-thread` f
 3 | #[tokio::main]
  | ^^^^^^^^^^^^^^
  |
-  = note: this error originates in the attribute macro `tokio::main` (in Nightly builds, run with -Z macro-backtrace for more info)
+  = note: this error originates in an attribute macro (in Nightly builds, run with -Z macro-backtrace for more info)
@@ -1,8 +0,0 @@
-#![deny(dead_code)]
-
-use tests_build::tokio;
-
-#[tokio::main]
-async fn f() {}
-
-fn main() {}
@@ -1,11 +0,0 @@
-error: function is never used: `f`
- --> $DIR/macros_dead_code.rs:6:10
-  |
-6 | async fn f() {}
-  |          ^
-  |
-note: the lint level is defined here
- --> $DIR/macros_dead_code.rs:1:9
-  |
-1 | #![deny(dead_code)]
-  |         ^^^^^^^^^
@@ -12,36 +12,12 @@ async fn main_attr_has_path_args() {}
 #[tokio::test]
 fn test_is_not_async() {}

+#[tokio::test]
+async fn test_fn_has_args(_x: u8) {}
+
 #[tokio::test(foo)]
 async fn test_attr_has_args() {}

-#[tokio::test(foo = 123)]
-async fn test_unexpected_attr() {}
-
-#[tokio::test(flavor = 123)]
-async fn test_flavor_not_string() {}
-
-#[tokio::test(flavor = "foo")]
-async fn test_unknown_flavor() {}
-
-#[tokio::test(flavor = "multi_thread", start_paused = false)]
-async fn test_multi_thread_with_start_paused() {}
-
-#[tokio::test(flavor = "multi_thread", worker_threads = "foo")]
-async fn test_worker_threads_not_int() {}
-
-#[tokio::test(flavor = "current_thread", worker_threads = 4)]
-async fn test_worker_threads_and_current_thread() {}
-
-#[tokio::test(crate = 456)]
-async fn test_crate_not_ident_int() {}
-
-#[tokio::test(crate = "456")]
-async fn test_crate_not_ident_invalid() {}
-
-#[tokio::test(crate = "abc::edf")]
-async fn test_crate_not_ident_path() {}
-
 #[tokio::test]
 #[test]
 async fn test_has_second_test_attr() {}
@@ -1,10 +1,10 @@
-error: the `async` keyword is missing from the function declaration
+error: the async keyword is missing from the function declaration
 --> $DIR/macros_invalid_input.rs:4:1
  |
 4 | fn main_is_not_async() {}
  | ^^

-error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`
+error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`
 --> $DIR/macros_invalid_input.rs:6:15
  |
 6 | #[tokio::main(foo)]
@@ -16,82 +16,26 @@ error: Must have specified ident
 9 | #[tokio::main(threadpool::bar)]
  |               ^^^^^^^^^^^^^^^

-error: the `async` keyword is missing from the function declaration
+error: the async keyword is missing from the function declaration
  --> $DIR/macros_invalid_input.rs:13:1
   |
 13 | fn test_is_not_async() {}
   | ^^

-error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`
-  --> $DIR/macros_invalid_input.rs:15:15
+error: the test function cannot accept arguments
+  --> $DIR/macros_invalid_input.rs:16:27
   |
-15 | #[tokio::test(foo)]
-   |               ^^^
+16 | async fn test_fn_has_args(_x: u8) {}
+   |                           ^^^^^^

-error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`
+error: Unknown attribute foo is specified; expected one of: `flavor`, `worker_threads`
  --> $DIR/macros_invalid_input.rs:18:15
   |
-18 | #[tokio::test(foo = 123)]
-   |               ^^^^^^^^^
-
-error: Failed to parse value of `flavor` as string.
-  --> $DIR/macros_invalid_input.rs:21:24
-   |
-21 | #[tokio::test(flavor = 123)]
-   |                        ^^^
-
-error: No such runtime flavor `foo`. The runtime flavors are `current_thread` and `multi_thread`.
-  --> $DIR/macros_invalid_input.rs:24:24
-   |
-24 | #[tokio::test(flavor = "foo")]
-   |                        ^^^^^
-
-error: The `start_paused` option requires the `current_thread` runtime flavor. Use `#[tokio::test(flavor = "current_thread")]`
-  --> $DIR/macros_invalid_input.rs:27:55
-   |
-27 | #[tokio::test(flavor = "multi_thread", start_paused = false)]
-   |                                                       ^^^^^
-
-error: Failed to parse value of `worker_threads` as integer.
-  --> $DIR/macros_invalid_input.rs:30:57
-   |
-30 | #[tokio::test(flavor = "multi_thread", worker_threads = "foo")]
-   |                                                         ^^^^^
-
-error: The `worker_threads` option requires the `multi_thread` runtime flavor. Use `#[tokio::test(flavor = "multi_thread")]`
-  --> $DIR/macros_invalid_input.rs:33:59
-   |
-33 | #[tokio::test(flavor = "current_thread", worker_threads = 4)]
-   |                                                           ^
-
-error: Failed to parse value of `crate` as ident.
-  --> $DIR/macros_invalid_input.rs:36:23
-   |
-36 | #[tokio::test(crate = 456)]
-   |                       ^^^
-
-error: Failed to parse value of `crate` as ident: "456"
-  --> $DIR/macros_invalid_input.rs:39:23
-   |
-39 | #[tokio::test(crate = "456")]
-   |                       ^^^^^
-
-error: Failed to parse value of `crate` as ident: "abc::edf"
-  --> $DIR/macros_invalid_input.rs:42:23
-   |
-42 | #[tokio::test(crate = "abc::edf")]
-   |                       ^^^^^^^^^^
+18 | #[tokio::test(foo)]
+   |               ^^^

 error: second test attribute is supplied
-  --> $DIR/macros_invalid_input.rs:46:1
+  --> $DIR/macros_invalid_input.rs:22:1
   |
-46 | #[test]
+22 | #[test]
   | ^^^^^^^
-
-error: duplicated attribute
-  --> $DIR/macros_invalid_input.rs:46:1
-   |
-46 | #[test]
-   | ^^^^^^^
-   |
-   = note: `-D duplicate-macro-attributes` implied by `-D warnings`
@@ -1,26 +0,0 @@
-use tests_build::tokio;
-
-#[tokio::main]
-async fn missing_semicolon_or_return_type() {
-    Ok(())
-}
-
-#[tokio::main]
-async fn missing_return_type() {
-    return Ok(());
-}
-
-#[tokio::main]
-async fn extra_semicolon() -> Result<(), ()> {
-    /* TODO(taiki-e): help message still wrong
-    help: try using a variant of the expected enum
-       |
-    23 |     Ok(Ok(());)
-       |
-    23 |     Err(Ok(());)
-       |
-    */
-    Ok(());
-}
-
-fn main() {}
@@ -1,40 +0,0 @@
-error[E0308]: mismatched types
- --> $DIR/macros_type_mismatch.rs:5:5
-  |
-4 | async fn missing_semicolon_or_return_type() {
-  |                                             - possibly return type missing here?
-5 |     Ok(())
-  |     ^^^^^^- help: consider using a semicolon here: `;`
-  |     |
-  |     expected `()`, found enum `Result`
-  |
-  = note: expected unit type `()`
-                  found enum `Result<(), _>`
-
-error[E0308]: mismatched types
-  --> $DIR/macros_type_mismatch.rs:10:5
-   |
-9  | async fn missing_return_type() {
-   |                                - possibly return type missing here?
-10 |     return Ok(());
-   |     ^^^^^^^^^^^^^^ expected `()`, found enum `Result`
-   |
-   = note: expected unit type `()`
-                   found enum `Result<(), _>`
-
-error[E0308]: mismatched types
-  --> $DIR/macros_type_mismatch.rs:23:5
-   |
-14 | async fn extra_semicolon() -> Result<(), ()> {
-   |                               -------------- expected `Result<(), ()>` because of return type
-...
-23 |     Ok(());
-   |     ^^^^^^^ expected enum `Result`, found `()`
-   |
-   = note:   expected enum `Result<(), ()>`
-           found unit type `()`
-help: try adding an expression at the end of the block
-   |
-23 ~     Ok(());;
-24 +     Ok(())
-   |
@@ -2,24 +2,9 @@
 fn compile_fail_full() {
    let t = trybuild::TestCases::new();

-    #[cfg(feature = "full")]
-    t.pass("tests/pass/forward_args_and_output.rs");
-
-    #[cfg(feature = "full")]
-    t.pass("tests/pass/macros_main_return.rs");
-
-    #[cfg(feature = "full")]
-    t.pass("tests/pass/macros_main_loop.rs");
-
    #[cfg(feature = "full")]
    t.compile_fail("tests/fail/macros_invalid_input.rs");

-    #[cfg(feature = "full")]
-    t.compile_fail("tests/fail/macros_dead_code.rs");
-
-    #[cfg(feature = "full")]
-    t.compile_fail("tests/fail/macros_type_mismatch.rs");
-
    #[cfg(all(feature = "rt", not(feature = "full")))]
    t.compile_fail("tests/fail/macros_core_no_default.rs");

@@ -1,7 +0,0 @@
-#[cfg(feature = "full")]
-#[tokio::test]
-async fn test_with_semicolon_without_return_type() {
-    #![deny(clippy::semicolon_if_nothing_returned)]
-
-    dbg!(0);
-}
@@ -1,13 +0,0 @@
-use tests_build::tokio;
-
-fn main() {}
-
-// arguments and output type is forwarded so other macros can access them
-
-#[tokio::test]
-async fn test_fn_has_args(_x: u8) {}
-
-#[tokio::test]
-async fn test_has_output() -> Result<(), Box<dyn std::error::Error>> {
-    Ok(())
-}
@@ -1,14 +0,0 @@
-use tests_build::tokio;
-
-#[tokio::main]
-async fn main() -> Result<(), ()> {
-    loop {
-        if !never() {
-            return Ok(());
-        }
-    }
-}
-
-fn never() -> bool {
-    std::time::Instant::now() > std::time::Instant::now()
-}
@@ -1,6 +0,0 @@
-use tests_build::tokio;
-
-#[tokio::main]
-async fn main() -> Result<(), ()> {
-    return Ok(());
-}
@@ -5,23 +5,7 @@ authors = ["Tokio Contributors <[email protected]>"]
 edition = "2018"
 publish = false

-[[bin]]
-name = "test-cat"
-
-[[bin]]
-name = "test-mem"
-required-features = ["rt-net"]
-
-[[bin]]
-name = "test-process-signal"
-required-features = ["rt-process-signal"]
-
 [features]
-# For mem check
-rt-net = ["tokio/rt", "tokio/rt-multi-thread", "tokio/net"]
-# For test-process-signal
-rt-process-signal = ["rt-net", "tokio/process", "tokio/signal"]
-
 full = [
    "macros",
    "rt",
@@ -39,4 +23,6 @@ rt-multi-thread = ["rt", "tokio/rt-multi-thread"]
 tokio = { path = "../tokio" }
 tokio-test = { path = "../tokio-test", optional = true }
 doc-comment = "0.3.1"
+
+[dev-dependencies]
 futures = { version = "0.3.0", features = ["async-await"] }
@@ -1,21 +0,0 @@
-use futures::future::poll_fn;
-
-fn main() {
-    let rt = tokio::runtime::Builder::new_multi_thread()
-        .worker_threads(1)
-        .enable_io()
-        .build()
-        .unwrap();
-
-    rt.block_on(async {
-        let listener = tokio::net::TcpListener::bind("0.0.0.0:0").await.unwrap();
-        tokio::spawn(async move {
-            loop {
-                poll_fn(|cx| listener.poll_accept(cx)).await.unwrap();
-            }
-        });
-    });
-
-    std::thread::sleep(std::time::Duration::from_millis(50));
-    drop(rt);
-}
@@ -1,11 +0,0 @@
-// https://github.com/tokio-rs/tokio/issues/3550
-fn main() {
-    for _ in 0..1000 {
-        let rt = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .unwrap();
-
-        drop(rt);
-    }
-}
@@ -1,19 +1,26 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]

-use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
-use tokio::join;
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::process::{Child, Command};
 use tokio_test::assert_ok;

 use futures::future::{self, FutureExt};
-use std::convert::TryInto;
 use std::env;
 use std::io;
 use std::process::{ExitStatus, Stdio};

 fn cat() -> Command {
-    let mut cmd = Command::new(env!("CARGO_BIN_EXE_test-cat"));
+    let mut me = env::current_exe().unwrap();
+    me.pop();
+
+    if me.ends_with("deps") {
+        me.pop();
+    }
+
+    me.push("test-cat");
+
+    let mut cmd = Command::new(me);
    cmd.stdin(Stdio::piped()).stdout(Stdio::piped());
    cmd
 }
@@ -141,52 +148,3 @@ async fn try_wait() {
    // Can't get id after process has exited
    assert_eq!(child.id(), None);
 }
-
-#[tokio::test]
-async fn pipe_from_one_command_to_another() {
-    let mut first = cat().spawn().expect("first cmd");
-    let mut third = cat().spawn().expect("third cmd");
-
-    // Convert ChildStdout to Stdio
-    let second_stdin: Stdio = first
-        .stdout
-        .take()
-        .expect("first.stdout")
-        .try_into()
-        .expect("first.stdout into Stdio");
-
-    // Convert ChildStdin to Stdio
-    let second_stdout: Stdio = third
-        .stdin
-        .take()
-        .expect("third.stdin")
-        .try_into()
-        .expect("third.stdin into Stdio");
-
-    let mut second = cat()
-        .stdin(second_stdin)
-        .stdout(second_stdout)
-        .spawn()
-        .expect("first cmd");
-
-    let msg = "hello world! please pipe this message through";
-
-    let mut stdin = first.stdin.take().expect("first.stdin");
-    let write = async move { stdin.write_all(msg.as_bytes()).await };
-
-    let mut stdout = third.stdout.take().expect("third.stdout");
-    let read = async move {
-        let mut data = String::new();
-        stdout.read_to_string(&mut data).await.map(|_| data)
-    };
-
-    let (read, write, first_status, second_status, third_status) =
-        join!(read, write, first.wait(), second.wait(), third.wait());
-
-    assert_eq!(msg, read.expect("read result"));
-    write.expect("write result");
-
-    assert!(first_status.expect("first status").success());
-    assert!(second_status.expect("second status").success());
-    assert!(third_status.expect("third status").success());
-}
@@ -1,71 +1,9 @@
-# 1.7.0 (December 15th, 2021)
-
- macros: address remainging clippy::semicolon_if_nothing_returned warning ([#4252])
-
-[#4252]: https://github.com/tokio-rs/tokio/pull/4252
-
-# 1.6.0 (November 16th, 2021)
-
- macros: fix mut patterns in `select!` macro ([#4211])
-
-[#4211]: https://github.com/tokio-rs/tokio/pull/4211
-
-# 1.5.1 (October 29th, 2021)
-
- macros: fix type resolution error in `#[tokio::main]` ([#4176])
-
-[#4176]: https://github.com/tokio-rs/tokio/pull/4176
-
-# 1.5.0 (October 13th, 2021)
-
- macros: make tokio-macros attributes more IDE friendly ([#4162])
-
-[#4162]: https://github.com/tokio-rs/tokio/pull/4162
-
-# 1.4.1 (September 30th, 2021)
-
-Reverted: run `current_thread` inside `LocalSet` ([#4027])
-
-# 1.4.0 (September 29th, 2021)
-
-(yanked)
-
-### Changed
-
- macros: run `current_thread` inside `LocalSet` ([#4027])
- macros: explicitly relaxed clippy lint for `.expect()` in runtime entry macro ([#4030])
+# 0.3.2 (December 19, 2020)

 ### Fixed

- macros: fix invalid error messages in functions wrapped with `#[main]` or `#[test]` ([#4067])
-
-[#4027]: https://github.com/tokio-rs/tokio/pull/4027
-[#4030]: https://github.com/tokio-rs/tokio/pull/4030
-[#4067]: https://github.com/tokio-rs/tokio/pull/4067
-
-# 1.3.0 (July 7, 2021)
-
- macros: don't trigger `clippy::unwrap_used` ([#3926])
-
-[#3926]: https://github.com/tokio-rs/tokio/pull/3926
-
-# 1.2.0 (May 14, 2021)
-
- macros: forward input arguments in `#[tokio::test]` ([#3691])
- macros: improve diagnostics on type mismatch ([#3766])
- macros: various error message improvements ([#3677])
-
-[#3677]: https://github.com/tokio-rs/tokio/pull/3677
-[#3691]: https://github.com/tokio-rs/tokio/pull/3691
-[#3766]: https://github.com/tokio-rs/tokio/pull/3766
-
-# 1.1.0 (February 5, 2021)
-
- add `start_paused` option to macros ([#3492])
-
-# 1.0.0 (December 23, 2020)
-
- track `tokio` 1.0 release.
+- fix outdated macro documentation ([#3180])
+- add portability note to `tokio::main` ([#3199])

 # 0.3.1 (October 25, 2020)

@@ -120,4 +58,5 @@ Reverted: run `current_thread` inside `LocalSet` ([#4027])
 [#2177]: https://github.com/tokio-rs/tokio/pull/2177
 [#2225]: https://github.com/tokio-rs/tokio/pull/2225
 [#3038]: https://github.com/tokio-rs/tokio/pull/3038
-[#3492]: https://github.com/tokio-rs/tokio/pull/3492
+[#3180]: https://github.com/tokio-rs/tokio/pull/3180
+[#3199]: https://github.com/tokio-rs/tokio/pull/3199
@@ -2,15 +2,18 @@
 name = "tokio-macros"
 # When releasing to crates.io:
 # - Remove path dependencies
+# - Update html_root_url.
+# - Update doc url
+#   - Cargo.toml
 # - Update CHANGELOG.md.
-# - Create "tokio-macros-1.0.x" git tag.
-version = "1.7.0"
+# - Create "v0.3.x" git tag.
+version = "0.3.2"
 edition = "2018"
-rust-version = "1.49"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
+documentation = "https://docs.rs/tokio-macros/0.3.2/tokio_macros"
 description = """
 Tokio's proc macros.
 """
@@ -24,10 +27,10 @@ proc-macro = true
 [dependencies]
 proc-macro2 = "1.0.7"
 quote = "1"
-syn = { version = "1.0.56", features = ["full"] }
+syn = { version = "1.0.3", features = ["full"] }

 [dev-dependencies]
-tokio = { version = "1.0.0", path = "../tokio", features = ["full"] }
+tokio = { version = "0.3.0", path = "../tokio", features = ["full"] }

 [package.metadata.docs.rs]
 all-features = true
@@ -1,4 +1,4 @@
-Copyright (c) 2022 Tokio Contributors
+Copyright (c) 2019 Tokio Contributors

 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated
@@ -1,10 +1,7 @@
 use proc_macro::TokenStream;
-use proc_macro2::{Ident, Span};
-use quote::{quote, quote_spanned, ToTokens};
-use syn::parse::Parser;
-
-// syn::AttributeArgs does not implement syn::Parse
-type AttributeArgs = syn::punctuated::Punctuated<syn::NestedMeta, syn::Token![,]>;
+use proc_macro2::Span;
+use quote::quote;
+use syn::spanned::Spanned;

 #[derive(Clone, Copy, PartialEq)]
 enum RuntimeFlavor {
@@ -28,26 +25,13 @@ impl RuntimeFlavor {
 struct FinalConfig {
    flavor: RuntimeFlavor,
    worker_threads: Option<usize>,
-    start_paused: Option<bool>,
-    crate_name: Option<String>,
 }

-/// Config used in case of the attribute not being able to build a valid config
-const DEFAULT_ERROR_CONFIG: FinalConfig = FinalConfig {
-    flavor: RuntimeFlavor::CurrentThread,
-    worker_threads: None,
-    start_paused: None,
-    crate_name: None,
-};
-
 struct Configuration {
    rt_multi_thread_available: bool,
    default_flavor: RuntimeFlavor,
    flavor: Option<RuntimeFlavor>,
    worker_threads: Option<(usize, Span)>,
-    start_paused: Option<(bool, Span)>,
-    is_test: bool,
-    crate_name: Option<String>,
 }

 impl Configuration {
@@ -60,9 +44,6 @@ impl Configuration {
            },
            flavor: None,
            worker_threads: None,
-            start_paused: None,
-            is_test,
-            crate_name: None,
        }
    }

@@ -98,77 +79,31 @@ impl Configuration {
        Ok(())
    }

-    fn set_start_paused(&mut self, start_paused: syn::Lit, span: Span) -> Result<(), syn::Error> {
-        if self.start_paused.is_some() {
-            return Err(syn::Error::new(span, "`start_paused` set multiple times."));
-        }
-
-        let start_paused = parse_bool(start_paused, span, "start_paused")?;
-        self.start_paused = Some((start_paused, span));
-        Ok(())
-    }
-
-    fn set_crate_name(&mut self, name: syn::Lit, span: Span) -> Result<(), syn::Error> {
-        if self.crate_name.is_some() {
-            return Err(syn::Error::new(span, "`crate` set multiple times."));
-        }
-        let name_ident = parse_ident(name, span, "crate")?;
-        self.crate_name = Some(name_ident.to_string());
-        Ok(())
-    }
-
-    fn macro_name(&self) -> &'static str {
-        if self.is_test {
-            "tokio::test"
-        } else {
-            "tokio::main"
-        }
-    }
-
    fn build(&self) -> Result<FinalConfig, syn::Error> {
        let flavor = self.flavor.unwrap_or(self.default_flavor);
        use RuntimeFlavor::*;
-
-        let worker_threads = match (flavor, self.worker_threads) {
-            (CurrentThread, Some((_, worker_threads_span))) => {
-                let msg = format!(
-                    "The `worker_threads` option requires the `multi_thread` runtime flavor. Use `#[{}(flavor = \"multi_thread\")]`",
-                    self.macro_name(),
-                );
-                return Err(syn::Error::new(worker_threads_span, msg));
-            }
-            (CurrentThread, None) => None,
-            (Threaded, worker_threads) if self.rt_multi_thread_available => {
-                worker_threads.map(|(val, _span)| val)
-            }
+        match (flavor, self.worker_threads) {
+            (CurrentThread, Some((_, worker_threads_span))) => Err(syn::Error::new(
+                worker_threads_span,
+                "The `worker_threads` option requires the `multi_thread` runtime flavor.",
+            )),
+            (CurrentThread, None) => Ok(FinalConfig {
+                flavor,
+                worker_threads: None,
+            }),
+            (Threaded, worker_threads) if self.rt_multi_thread_available => Ok(FinalConfig {
+                flavor,
+                worker_threads: worker_threads.map(|(val, _span)| val),
+            }),
            (Threaded, _) => {
                let msg = if self.flavor.is_none() {
                    "The default runtime flavor is `multi_thread`, but the `rt-multi-thread` feature is disabled."
                } else {
                    "The runtime flavor `multi_thread` requires the `rt-multi-thread` feature."
                };
-                return Err(syn::Error::new(Span::call_site(), msg));
+                Err(syn::Error::new(Span::call_site(), msg))
            }
-        };
-
-        let start_paused = match (flavor, self.start_paused) {
-            (Threaded, Some((_, start_paused_span))) => {
-                let msg = format!(
-                    "The `start_paused` option requires the `current_thread` runtime flavor. Use `#[{}(flavor = \"current_thread\")]`",
-                    self.macro_name(),
-                );
-                return Err(syn::Error::new(start_paused_span, msg));
-            }
-            (CurrentThread, Some((start_paused, _))) => Some(start_paused),
-            (_, None) => None,
-        };
-
-        Ok(FinalConfig {
-            crate_name: self.crate_name.clone(),
-            flavor,
-            worker_threads,
-            start_paused,
-        })
+        }
    }
 }

@@ -178,12 +113,12 @@ fn parse_int(int: syn::Lit, span: Span, field: &str) -> Result<usize, syn::Error
            Ok(value) => Ok(value),
            Err(e) => Err(syn::Error::new(
                span,
-                format!("Failed to parse value of `{}` as integer: {}", field, e),
+                format!("Failed to parse {} as integer: {}", field, e),
            )),
        },
        _ => Err(syn::Error::new(
            span,
-            format!("Failed to parse value of `{}` as integer.", field),
+            format!("Failed to parse {} as integer.", field),
        )),
    }
 }
@@ -194,111 +129,68 @@ fn parse_string(int: syn::Lit, span: Span, field: &str) -> Result<String, syn::E
        syn::Lit::Verbatim(s) => Ok(s.to_string()),
        _ => Err(syn::Error::new(
            span,
-            format!("Failed to parse value of `{}` as string.", field),
+            format!("Failed to parse {} as string.", field),
        )),
    }
 }

-fn parse_ident(lit: syn::Lit, span: Span, field: &str) -> Result<Ident, syn::Error> {
-    match lit {
-        syn::Lit::Str(s) => {
-            let err = syn::Error::new(
-                span,
-                format!(
-                    "Failed to parse value of `{}` as ident: \"{}\"",
-                    field,
-                    s.value()
-                ),
-            );
-            let path = s.parse::<syn::Path>().map_err(|_| err.clone())?;
-            path.get_ident().cloned().ok_or(err)
-        }
-        _ => Err(syn::Error::new(
-            span,
-            format!("Failed to parse value of `{}` as ident.", field),
-        )),
-    }
-}
-
-fn parse_bool(bool: syn::Lit, span: Span, field: &str) -> Result<bool, syn::Error> {
-    match bool {
-        syn::Lit::Bool(b) => Ok(b.value),
-        _ => Err(syn::Error::new(
-            span,
-            format!("Failed to parse value of `{}` as bool.", field),
-        )),
-    }
-}
-
-fn build_config(
-    input: syn::ItemFn,
-    args: AttributeArgs,
+fn parse_knobs(
+    mut input: syn::ItemFn,
+    args: syn::AttributeArgs,
    is_test: bool,
    rt_multi_thread: bool,
-) -> Result<FinalConfig, syn::Error> {
-    if input.sig.asyncness.is_none() {
-        let msg = "the `async` keyword is missing from the function declaration";
-        return Err(syn::Error::new_spanned(input.sig.fn_token, msg));
+) -> Result<TokenStream, syn::Error> {
+    let sig = &mut input.sig;
+    let body = &input.block;
+    let attrs = &input.attrs;
+    let vis = input.vis;
+
+    if sig.asyncness.is_none() {
+        let msg = "the async keyword is missing from the function declaration";
+        return Err(syn::Error::new_spanned(sig.fn_token, msg));
    }

+    sig.asyncness = None;
+
+    let macro_name = if is_test {
+        "tokio::test"
+    } else {
+        "tokio::main"
+    };
    let mut config = Configuration::new(is_test, rt_multi_thread);
-    let macro_name = config.macro_name();

    for arg in args {
        match arg {
            syn::NestedMeta::Meta(syn::Meta::NameValue(namevalue)) => {
-                let ident = namevalue
-                    .path
-                    .get_ident()
-                    .ok_or_else(|| {
-                        syn::Error::new_spanned(&namevalue, "Must have specified ident")
-                    })?
-                    .to_string()
-                    .to_lowercase();
-                match ident.as_str() {
+                let ident = namevalue.path.get_ident();
+                if ident.is_none() {
+                    let msg = "Must have specified ident";
+                    return Err(syn::Error::new_spanned(namevalue, msg));
+                }
+                match ident.unwrap().to_string().to_lowercase().as_str() {
                    "worker_threads" => {
-                        config.set_worker_threads(
-                            namevalue.lit.clone(),
-                            syn::spanned::Spanned::span(&namevalue.lit),
-                        )?;
+                        config.set_worker_threads(namevalue.lit.clone(), namevalue.span())?;
                    }
                    "flavor" => {
-                        config.set_flavor(
-                            namevalue.lit.clone(),
-                            syn::spanned::Spanned::span(&namevalue.lit),
-                        )?;
-                    }
-                    "start_paused" => {
-                        config.set_start_paused(
-                            namevalue.lit.clone(),
-                            syn::spanned::Spanned::span(&namevalue.lit),
-                        )?;
+                        config.set_flavor(namevalue.lit.clone(), namevalue.span())?;
                    }
                    "core_threads" => {
                        let msg = "Attribute `core_threads` is renamed to `worker_threads`";
                        return Err(syn::Error::new_spanned(namevalue, msg));
                    }
-                    "crate" => {
-                        config.set_crate_name(
-                            namevalue.lit.clone(),
-                            syn::spanned::Spanned::span(&namevalue.lit),
-                        )?;
-                    }
                    name => {
-                        let msg = format!(
-                            "Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`",
-                            name,
-                        );
+                        let msg = format!("Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`", name);
                        return Err(syn::Error::new_spanned(namevalue, msg));
                    }
                }
            }
            syn::NestedMeta::Meta(syn::Meta::Path(path)) => {
-                let name = path
-                    .get_ident()
-                    .ok_or_else(|| syn::Error::new_spanned(&path, "Must have specified ident"))?
-                    .to_string()
-                    .to_lowercase();
+                let ident = path.get_ident();
+                if ident.is_none() {
+                    let msg = "Must have specified ident";
+                    return Err(syn::Error::new_spanned(path, msg));
+                }
+                let name = ident.unwrap().to_string().to_lowercase();
                let msg = match name.as_str() {
                    "threaded_scheduler" | "multi_thread" => {
                        format!(
@@ -312,11 +204,11 @@ fn build_config(
                            macro_name
                        )
                    }
-                    "flavor" | "worker_threads" | "start_paused" => {
+                    "flavor" | "worker_threads" => {
                        format!("The `{}` attribute requires an argument.", name)
                    }
                    name => {
-                        format!("Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`, `start_paused`, `crate`", name)
+                        format!("Unknown attribute {} is specified; expected one of: `flavor`, `worker_threads`", name)
                    }
                };
                return Err(syn::Error::new_spanned(path, msg));
@@ -330,144 +222,79 @@ fn build_config(
        }
    }

-    config.build()
-}
-
-fn parse_knobs(mut input: syn::ItemFn, is_test: bool, config: FinalConfig) -> TokenStream {
-    input.sig.asyncness = None;
-
-    // If type mismatch occurs, the current rustc points to the last statement.
-    let (last_stmt_start_span, last_stmt_end_span) = {
-        let mut last_stmt = input
-            .block
-            .stmts
-            .last()
-            .map(ToTokens::into_token_stream)
-            .unwrap_or_default()
-            .into_iter();
-        // `Span` on stable Rust has a limitation that only points to the first
-        // token, not the whole tokens. We can work around this limitation by
-        // using the first/last span of the tokens like
-        // `syn::Error::new_spanned` does.
-        let start = last_stmt.next().map_or_else(Span::call_site, |t| t.span());
-        let end = last_stmt.last().map_or(start, |t| t.span());
-        (start, end)
-    };
-
-    let crate_name = config.crate_name.as_deref().unwrap_or("tokio");
-
-    let crate_ident = Ident::new(crate_name, last_stmt_start_span);
+    let config = config.build()?;

    let mut rt = match config.flavor {
-        RuntimeFlavor::CurrentThread => quote_spanned! {last_stmt_start_span=>
-            #crate_ident::runtime::Builder::new_current_thread()
+        RuntimeFlavor::CurrentThread => quote! {
+            tokio::runtime::Builder::new_current_thread()
        },
-        RuntimeFlavor::Threaded => quote_spanned! {last_stmt_start_span=>
-            #crate_ident::runtime::Builder::new_multi_thread()
+        RuntimeFlavor::Threaded => quote! {
+            tokio::runtime::Builder::new_multi_thread()
        },
    };
    if let Some(v) = config.worker_threads {
        rt = quote! { #rt.worker_threads(#v) };
    }
-    if let Some(v) = config.start_paused {
-        rt = quote! { #rt.start_paused(#v) };
-    }

-    let header = if is_test {
-        quote! {
-            #[::core::prelude::v1::test]
-        }
-    } else {
-        quote! {}
-    };
-
-    let body = &input.block;
-    let brace_token = input.block.brace_token;
-    let (tail_return, tail_semicolon) = match body.stmts.last() {
-        Some(syn::Stmt::Semi(syn::Expr::Return(_), _)) => (quote! { return }, quote! { ; }),
-        Some(syn::Stmt::Semi(..)) | Some(syn::Stmt::Local(..)) | None => {
-            match &input.sig.output {
-                syn::ReturnType::Type(_, ty) if matches!(&**ty, syn::Type::Tuple(ty) if ty.elems.is_empty()) =>
-                {
-                    (quote! {}, quote! { ; }) // unit
-                }
-                syn::ReturnType::Default => (quote! {}, quote! { ; }), // unit
-                syn::ReturnType::Type(..) => (quote! {}, quote! {}),   // ! or another
+    let header = {
+        if is_test {
+            quote! {
+                #[::core::prelude::v1::test]
            }
+        } else {
+            quote! {}
        }
-        _ => (quote! {}, quote! {}),
    };
-    input.block = syn::parse2(quote_spanned! {last_stmt_end_span=>
-        {
-            let body = async #body;
-            #[allow(clippy::expect_used)]
-            #tail_return #rt
-                .enable_all()
-                .build()
-                .expect("Failed building the Runtime")
-                .block_on(body)#tail_semicolon
-        }
-    })
-    .expect("Parsing failure");
-    input.block.brace_token = brace_token;

    let result = quote! {
        #header
-        #input
+        #(#attrs)*
+        #vis #sig {
+            #rt
+                .enable_all()
+                .build()
+                .unwrap()
+                .block_on(async { #body })
+        }
    };

-    result.into()
-}
-
-fn token_stream_with_error(mut tokens: TokenStream, error: syn::Error) -> TokenStream {
-    tokens.extend(TokenStream::from(error.into_compile_error()));
-    tokens
+    Ok(result.into())
 }

 #[cfg(not(test))] // Work around for rust-lang/rust#62127
 pub(crate) fn main(args: TokenStream, item: TokenStream, rt_multi_thread: bool) -> TokenStream {
-    // If any of the steps for this macro fail, we still want to expand to an item that is as close
-    // to the expected output as possible. This helps out IDEs such that completions and other
-    // related features keep working.
-    let input: syn::ItemFn = match syn::parse(item.clone()) {
-        Ok(it) => it,
-        Err(e) => return token_stream_with_error(item, e),
-    };
+    let input = syn::parse_macro_input!(item as syn::ItemFn);
+    let args = syn::parse_macro_input!(args as syn::AttributeArgs);

-    let config = if input.sig.ident == "main" && !input.sig.inputs.is_empty() {
+    if input.sig.ident == "main" && !input.sig.inputs.is_empty() {
        let msg = "the main function cannot accept arguments";
-        Err(syn::Error::new_spanned(&input.sig.ident, msg))
-    } else {
-        AttributeArgs::parse_terminated
-            .parse(args)
-            .and_then(|args| build_config(input.clone(), args, false, rt_multi_thread))
-    };
-
-    match config {
-        Ok(config) => parse_knobs(input, false, config),
-        Err(e) => token_stream_with_error(parse_knobs(input, false, DEFAULT_ERROR_CONFIG), e),
+        return syn::Error::new_spanned(&input.sig.ident, msg)
+            .to_compile_error()
+            .into();
    }
+
+    parse_knobs(input, args, false, rt_multi_thread).unwrap_or_else(|e| e.to_compile_error().into())
 }

 pub(crate) fn test(args: TokenStream, item: TokenStream, rt_multi_thread: bool) -> TokenStream {
-    // If any of the steps for this macro fail, we still want to expand to an item that is as close
-    // to the expected output as possible. This helps out IDEs such that completions and other
-    // related features keep working.
-    let input: syn::ItemFn = match syn::parse(item.clone()) {
-        Ok(it) => it,
-        Err(e) => return token_stream_with_error(item, e),
-    };
-    let config = if let Some(attr) = input.attrs.iter().find(|attr| attr.path.is_ident("test")) {
-        let msg = "second test attribute is supplied";
-        Err(syn::Error::new_spanned(&attr, msg))
-    } else {
-        AttributeArgs::parse_terminated
-            .parse(args)
-            .and_then(|args| build_config(input.clone(), args, true, rt_multi_thread))
-    };
+    let input = syn::parse_macro_input!(item as syn::ItemFn);
+    let args = syn::parse_macro_input!(args as syn::AttributeArgs);

-    match config {
-        Ok(config) => parse_knobs(input, true, config),
-        Err(e) => token_stream_with_error(parse_knobs(input, true, DEFAULT_ERROR_CONFIG), e),
+    for attr in &input.attrs {
+        if attr.path.is_ident("test") {
+            let msg = "second test attribute is supplied";
+            return syn::Error::new_spanned(&attr, msg)
+                .to_compile_error()
+                .into();
+        }
    }
+
+    if !input.sig.inputs.is_empty() {
+        let msg = "the test function cannot accept arguments";
+        return syn::Error::new_spanned(&input.sig.inputs, msg)
+            .to_compile_error()
+            .into();
+    }
+
+    parse_knobs(input, args, true, rt_multi_thread).unwrap_or_else(|e| e.to_compile_error().into())
 }
@@ -1,3 +1,4 @@
+#![doc(html_root_url = "https://docs.rs/tokio-macros/0.3.2")]
 #![allow(clippy::needless_doctest_main)]
 #![warn(
    missing_debug_implementations,
@@ -5,6 +6,7 @@
    rust_2018_idioms,
    unreachable_pub
 )]
+#![cfg_attr(docsrs, deny(broken_intra_doc_links))]
 #![doc(test(
    no_crate_inject,
    attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))
@@ -22,8 +24,8 @@ mod select;

 use proc_macro::TokenStream;

-/// Marks async function to be executed by the selected runtime. This macro
-/// helps set up a `Runtime` without requiring the user to use
+/// Marks async function to be executed by the selected runtime. This macro helps
+/// set up a `Runtime` without requiring the user to use
 /// [Runtime](../tokio/runtime/struct.Runtime.html) or
 /// [Builder](../tokio/runtime/struct.Builder.html) directly.
 ///
@@ -34,10 +36,10 @@ use proc_macro::TokenStream;
 /// powerful interface.
 ///
 /// Note: This macro can be used on any function and not just the `main`
-/// function. Using it on a non-main function makes the function behave as if it
-/// was synchronous by starting a new runtime each time it is called. If the
-/// function is called often, it is preferable to create the runtime using the
-/// runtime builder so the runtime can be reused across calls.
+/// function. Using it on a non-main function makes the function behave
+/// as if it was synchronous by starting a new runtime each time it is called.
+/// If the function is called often, it is preferable to create the runtime using
+/// the runtime builder so the runtime can be reused across calls.
 ///
 /// # Multi-threaded runtime
 ///
@@ -51,9 +53,6 @@ use proc_macro::TokenStream;
 /// The `worker_threads` option configures the number of worker threads, and
 /// defaults to the number of cpus on the system. This is the default flavor.
 ///
-/// Note: The multi-threaded runtime requires the `rt-multi-thread` feature
-/// flag.
-///
 /// # Current thread runtime
 ///
 /// To use the single-threaded runtime known as the `current_thread` runtime,
@@ -142,58 +141,12 @@ use proc_macro::TokenStream;
 /// }
 /// ```
 ///
-/// ### Configure the runtime to start with time paused
+/// ### NOTE:
 ///
-/// ```rust
-/// #[tokio::main(flavor = "current_thread", start_paused = true)]
-/// async fn main() {
-///     println!("Hello world");
-/// }
-/// ```
-///
-/// Equivalent code not using `#[tokio::main]`
-///
-/// ```rust
-/// fn main() {
-///     tokio::runtime::Builder::new_current_thread()
-///         .enable_all()
-///         .start_paused(true)
-///         .build()
-///         .unwrap()
-///         .block_on(async {
-///             println!("Hello world");
-///         })
-/// }
-/// ```
-///
-/// Note that `start_paused` requires the `test-util` feature to be enabled.
-///
-/// ### Rename package
-///
-/// ```rust
-/// use tokio as tokio1;
-///
-/// #[tokio1::main(crate = "tokio1")]
-/// async fn main() {
-///     println!("Hello world");
-/// }
-/// ```
-///
-/// Equivalent code not using `#[tokio::main]`
-///
-/// ```rust
-/// use tokio as tokio1;
-///
-/// fn main() {
-///     tokio1::runtime::Builder::new_multi_thread()
-///         .enable_all()
-///         .build()
-///         .unwrap()
-///         .block_on(async {
-///             println!("Hello world");
-///         })
-/// }
-/// ```
+/// If you rename the tokio crate in your dependencies this macro will not work.
+/// If you must rename the 0.3 version of tokio because you're also using the
+/// 0.1 version of tokio, you _must_ make the tokio 0.3 crate available as
+/// `tokio` in the module where this macro is expanded.
 #[proc_macro_attribute]
 #[cfg(not(test))] // Work around for rust-lang/rust#62127
 pub fn main(args: TokenStream, item: TokenStream) -> TokenStream {
@@ -233,32 +186,13 @@ pub fn main(args: TokenStream, item: TokenStream) -> TokenStream {
 /// }
 /// ```
 ///
-/// ### Rename package
+/// ### NOTE:
 ///
-/// ```rust
-/// use tokio as tokio1;
-///
-/// #[tokio1::main(crate = "tokio1")]
-/// async fn main() {
-///     println!("Hello world");
-/// }
-/// ```
-///
-/// Equivalent code not using `#[tokio::main]`
-///
-/// ```rust
-/// use tokio as tokio1;
-///
-/// fn main() {
-///     tokio1::runtime::Builder::new_multi_thread()
-///         .enable_all()
-///         .build()
-///         .unwrap()
-///         .block_on(async {
-///             println!("Hello world");
-///         })
-/// }
-/// ```
+/// If you rename the tokio crate in your dependencies this macro
+/// will not work. If you must rename the 0.3 version of tokio because
+/// you're also using the 0.1 version of tokio, you _must_ make the
+/// tokio 0.3 crate available as `tokio` in the module where this
+/// macro is expanded.
 #[proc_macro_attribute]
 #[cfg(not(test))] // Work around for rust-lang/rust#62127
 pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
@@ -289,27 +223,13 @@ pub fn main_rt(args: TokenStream, item: TokenStream) -> TokenStream {
 /// }
 /// ```
 ///
-/// ### Configure the runtime to start with time paused
+/// ### NOTE:
 ///
-/// ```no_run
-/// #[tokio::test(start_paused = true)]
-/// async fn my_test() {
-///     assert!(true);
-/// }
-/// ```
-///
-/// Note that `start_paused` requires the `test-util` feature to be enabled.
-///
-/// ### Rename package
-///
-/// ```rust
-/// use tokio as tokio1;
-///
-/// #[tokio1::test(crate = "tokio1")]
-/// async fn my_test() {
-///     println!("Hello world");
-/// }
-/// ```
+/// If you rename the tokio crate in your dependencies this macro
+/// will not work. If you must rename the 0.3 version of tokio because
+/// you're also using the 0.1 version of tokio, you _must_ make the
+/// tokio 0.3 crate available as `tokio` in the module where this
+/// macro is expanded.
 #[proc_macro_attribute]
 pub fn test(args: TokenStream, item: TokenStream) -> TokenStream {
    entry::test(args, item, true)
@@ -325,6 +245,14 @@ pub fn test(args: TokenStream, item: TokenStream) -> TokenStream {
 ///     assert!(true);
 /// }
 /// ```
+///
+/// ### NOTE:
+///
+/// If you rename the tokio crate in your dependencies this macro
+/// will not work. If you must rename the 0.3 version of tokio because
+/// you're also using the 0.1 version of tokio, you _must_ make the
+/// tokio 0.3 crate available as `tokio` in the module where this
+/// macro is expanded.
 #[proc_macro_attribute]
 pub fn test_rt(args: TokenStream, item: TokenStream) -> TokenStream {
    entry::test(args, item, false)
@@ -365,11 +293,3 @@ pub fn test_fail(_args: TokenStream, _item: TokenStream) -> TokenStream {
 pub fn select_priv_declare_output_enum(input: TokenStream) -> TokenStream {
    select::declare_output_enum(input)
 }
-
-/// Implementation detail of the `select!` macro. This macro is **not** intended
-/// to be used as part of the public API and is permitted to change.
-#[proc_macro]
-#[doc(hidden)]
-pub fn select_priv_clean_pattern(input: TokenStream) -> TokenStream {
-    select::clean_pattern_macro(input)
-}
@@ -41,70 +41,3 @@ pub(crate) fn declare_output_enum(input: TokenStream) -> TokenStream {
        pub(super) type Mask = #mask;
    })
 }
-
-pub(crate) fn clean_pattern_macro(input: TokenStream) -> TokenStream {
-    // If this isn't a pattern, we return the token stream as-is. The select!
-    // macro is using it in a location requiring a pattern, so an error will be
-    // emitted there.
-    let mut input: syn::Pat = match syn::parse(input.clone()) {
-        Ok(it) => it,
-        Err(_) => return input,
-    };
-
-    clean_pattern(&mut input);
-    quote::ToTokens::into_token_stream(input).into()
-}
-
-// Removes any occurrences of ref or mut in the provided pattern.
-fn clean_pattern(pat: &mut syn::Pat) {
-    match pat {
-        syn::Pat::Box(_box) => {}
-        syn::Pat::Lit(_literal) => {}
-        syn::Pat::Macro(_macro) => {}
-        syn::Pat::Path(_path) => {}
-        syn::Pat::Range(_range) => {}
-        syn::Pat::Rest(_rest) => {}
-        syn::Pat::Verbatim(_tokens) => {}
-        syn::Pat::Wild(_underscore) => {}
-        syn::Pat::Ident(ident) => {
-            ident.by_ref = None;
-            ident.mutability = None;
-            if let Some((_at, pat)) = &mut ident.subpat {
-                clean_pattern(&mut *pat);
-            }
-        }
-        syn::Pat::Or(or) => {
-            for case in or.cases.iter_mut() {
-                clean_pattern(case);
-            }
-        }
-        syn::Pat::Slice(slice) => {
-            for elem in slice.elems.iter_mut() {
-                clean_pattern(elem);
-            }
-        }
-        syn::Pat::Struct(struct_pat) => {
-            for field in struct_pat.fields.iter_mut() {
-                clean_pattern(&mut field.pat);
-            }
-        }
-        syn::Pat::Tuple(tuple) => {
-            for elem in tuple.elems.iter_mut() {
-                clean_pattern(elem);
-            }
-        }
-        syn::Pat::TupleStruct(tuple) => {
-            for elem in tuple.pat.elems.iter_mut() {
-                clean_pattern(elem);
-            }
-        }
-        syn::Pat::Reference(reference) => {
-            reference.mutability = None;
-            clean_pattern(&mut *reference.pat);
-        }
-        syn::Pat::Type(type_pat) => {
-            clean_pattern(&mut *type_pat.pat);
-        }
-        _ => {}
-    }
-}
@@ -1,89 +0,0 @@
-# 0.1.8 (October 29, 2021)
-
- stream: add `From<Receiver<T>>` impl for receiver streams ([#4080])
- stream: impl `FromIterator` for `StreamMap` ([#4052])
- signal: make windows docs for signal module show up on unix builds ([#3770])
-
-[#3770]: https://github.com/tokio-rs/tokio/pull/3770
-[#4052]: https://github.com/tokio-rs/tokio/pull/4052
-[#4080]: https://github.com/tokio-rs/tokio/pull/4080
-
-# 0.1.7 (July 7, 2021)
-
-### Fixed
-
- sync: fix watch wrapper ([#3914])
- time: fix `Timeout::size_hint` ([#3902])
-
-[#3902]: https://github.com/tokio-rs/tokio/pull/3902
-[#3914]: https://github.com/tokio-rs/tokio/pull/3914
-
-# 0.1.6 (May 14, 2021)
-
-### Added
-
- stream: implement `Error` and `Display` for `BroadcastStreamRecvError` ([#3745])
-
-### Fixed
-
- stream: avoid yielding in `AllFuture` and `AnyFuture` ([#3625])
-
-[#3745]: https://github.com/tokio-rs/tokio/pull/3745
-[#3625]: https://github.com/tokio-rs/tokio/pull/3625
-
-# 0.1.5 (March 20, 2021)
-
-### Fixed
-
- stream: documentation note for throttle `Unpin` ([#3600])
-
-[#3600]: https://github.com/tokio-rs/tokio/pull/3600
-
-# 0.1.4 (March 9, 2021)
-
-Added
-
- signal: add `Signal` wrapper ([#3510])
-
-Fixed
-
- stream: remove duplicate `doc_cfg` declaration ([#3561])
- sync: yield initial value in `WatchStream` ([#3576])
-
-[#3510]: https://github.com/tokio-rs/tokio/pull/3510
-[#3561]: https://github.com/tokio-rs/tokio/pull/3561
-[#3576]: https://github.com/tokio-rs/tokio/pull/3576
-
-# 0.1.3 (February 5, 2021)
-
-Added
-
- - sync: add wrapper for broadcast and watch ([#3384], [#3504])
-
-[#3384]: https://github.com/tokio-rs/tokio/pull/3384
-[#3504]: https://github.com/tokio-rs/tokio/pull/3504
-
-# 0.1.2 (January 12, 2021)
-
-Fixed
-
- - docs: fix some wrappers missing in documentation ([#3378])
-
-[#3378]: https://github.com/tokio-rs/tokio/pull/3378
-
-# 0.1.1 (January 4, 2021)
-
-Added
-
- - add `Stream` wrappers ([#3343])
-
-Fixed
-
- - move `async-stream` to `dev-dependencies` ([#3366])
-
-[#3366]: https://github.com/tokio-rs/tokio/pull/3366
-[#3343]: https://github.com/tokio-rs/tokio/pull/3343
-
-# 0.1.0 (December 23, 2020)
-
- - Initial release
@@ -1,49 +0,0 @@
-[package]
-name = "tokio-stream"
-# When releasing to crates.io:
-# - Remove path dependencies
-# - Update CHANGELOG.md.
-# - Create "tokio-stream-0.1.x" git tag.
-version = "0.1.8"
-edition = "2018"
-rust-version = "1.49"
-authors = ["Tokio Contributors <[email protected]>"]
-license = "MIT"
-repository = "https://github.com/tokio-rs/tokio"
-homepage = "https://tokio.rs"
-description = """
-Utilities to work with `Stream` and `tokio`.
-"""
-categories = ["asynchronous"]
-
-[features]
-default = ["time"]
-time = ["tokio/time"]
-net = ["tokio/net"]
-io-util = ["tokio/io-util"]
-fs = ["tokio/fs"]
-sync = ["tokio/sync", "tokio-util"]
-signal = ["tokio/signal"]
-
-[dependencies]
-futures-core = { version = "0.3.0" }
-pin-project-lite = "0.2.0"
-tokio = { version = "1.8.0", path = "../tokio", features = ["sync"] }
-tokio-util = { version = "0.7.0", path = "../tokio-util", optional = true }
-
-[dev-dependencies]
-tokio = { version = "1.2.0", path = "../tokio", features = ["full", "test-util"] }
-async-stream = "0.3"
-tokio-test = { path = "../tokio-test" }
-futures = { version = "0.3", default-features = false }
-
-proptest = "1"
-
-[package.metadata.docs.rs]
-all-features = true
-rustdoc-args = ["--cfg", "docsrs"]
-# Issue #3770
-#
-# This should allow `docsrs` to be read across projects, so that `tokio-stream`
-# can pick up stubbed types exported by `tokio`.
-rustc-args = ["--cfg", "docsrs"]
@@ -1,25 +0,0 @@
-Copyright (c) 2022 Tokio Contributors
-
-Permission is hereby granted, free of charge, to any
-person obtaining a copy of this software and associated
-documentation files (the "Software"), to deal in the
-Software without restriction, including without
-limitation the rights to use, copy, modify, merge,
-publish, distribute, sublicense, and/or sell copies of
-the Software, and to permit persons to whom the Software
-is furnished to do so, subject to the following
-conditions:
-
-The above copyright notice and this permission notice
-shall be included in all copies or substantial portions
-of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
-ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
-TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
-PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
-SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
-CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
-OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
-IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
-DEALINGS IN THE SOFTWARE.
@@ -1,100 +0,0 @@
-#![allow(
-    clippy::cognitive_complexity,
-    clippy::large_enum_variant,
-    clippy::needless_doctest_main
-)]
-#![warn(
-    missing_debug_implementations,
-    missing_docs,
-    rust_2018_idioms,
-    unreachable_pub
-)]
-#![cfg_attr(docsrs, feature(doc_cfg))]
-#![doc(test(
-    no_crate_inject,
-    attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))
-))]
-
-//! Stream utilities for Tokio.
-//!
-//! A `Stream` is an asynchronous sequence of values. It can be thought of as
-//! an asynchronous version of the standard library's `Iterator` trait.
-//!
-//! This crate provides helpers to work with them. For examples of usage and a more in-depth
-//! description of streams you can also refer to the [streams
-//! tutorial](https://tokio.rs/tokio/tutorial/streams) on the tokio website.
-//!
-//! # Iterating over a Stream
-//!
-//! Due to similarities with the standard library's `Iterator` trait, some new
-//! users may assume that they can use `for in` syntax to iterate over a
-//! `Stream`, but this is unfortunately not possible. Instead, you can use a
-//! `while let` loop as follows:
-//!
-//! ```rust
-//! use tokio_stream::{self as stream, StreamExt};
-//!
-//! #[tokio::main]
-//! async fn main() {
-//!     let mut stream = stream::iter(vec![0, 1, 2]);
-//!
-//!     while let Some(value) = stream.next().await {
-//!         println!("Got {}", value);
-//!     }
-//! }
-//! ```
-//!
-//! # Returning a Stream from a function
-//!
-//! A common way to stream values from a function is to pass in the sender
-//! half of a channel and use the receiver as the stream. This requires awaiting
-//! both futures to ensure progress is made. Another alternative is the
-//! [async-stream] crate, which contains macros that provide a `yield` keyword
-//! and allow you to return an `impl Stream`.
-//!
-//! [async-stream]: https://docs.rs/async-stream
-//!
-//! # Conversion to and from AsyncRead/AsyncWrite
-//!
-//! It is often desirable to convert a `Stream` into an [`AsyncRead`],
-//! especially when dealing with plaintext formats streamed over the network.
-//! The opposite conversion from an [`AsyncRead`] into a `Stream` is also
-//! another commonly required feature. To enable these conversions,
-//! [`tokio-util`] provides the [`StreamReader`] and [`ReaderStream`]
-//! types when the io feature is enabled.
-//!
-//! [`tokio-util`]: https://docs.rs/tokio-util/0.4/tokio_util/codec/index.html
-//! [`tokio::io`]: https://docs.rs/tokio/1.0/tokio/io/index.html
-//! [`AsyncRead`]: https://docs.rs/tokio/1.0/tokio/io/trait.AsyncRead.html
-//! [`AsyncWrite`]: https://docs.rs/tokio/1.0/tokio/io/trait.AsyncWrite.html
-//! [`ReaderStream`]: https://docs.rs/tokio-util/0.4/tokio_util/io/struct.ReaderStream.html
-//! [`StreamReader`]: https://docs.rs/tokio-util/0.4/tokio_util/io/struct.StreamReader.html
-
-#[macro_use]
-mod macros;
-
-pub mod wrappers;
-
-mod stream_ext;
-pub use stream_ext::{collect::FromStream, StreamExt};
-cfg_time! {
-    pub use stream_ext::timeout::{Elapsed, Timeout};
-}
-
-mod empty;
-pub use empty::{empty, Empty};
-
-mod iter;
-pub use iter::{iter, Iter};
-
-mod once;
-pub use once::{once, Once};
-
-mod pending;
-pub use pending::{pending, Pending};
-
-mod stream_map;
-pub use stream_map::StreamMap;
-
-#[doc(no_inline)]
-pub use futures_core::Stream;
@@ -1,68 +0,0 @@
-macro_rules! cfg_fs {
-    ($($item:item)*) => {
-        $(
-            #[cfg(feature = "fs")]
-            #[cfg_attr(docsrs, doc(cfg(feature = "fs")))]
-            $item
-        )*
-    }
-}
-
-macro_rules! cfg_io_util {
-    ($($item:item)*) => {
-        $(
-            #[cfg(feature = "io-util")]
-            #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
-            $item
-        )*
-    }
-}
-
-macro_rules! cfg_net {
-    ($($item:item)*) => {
-        $(
-            #[cfg(feature = "net")]
-            #[cfg_attr(docsrs, doc(cfg(feature = "net")))]
-            $item
-        )*
-    }
-}
-
-macro_rules! cfg_time {
-    ($($item:item)*) => {
-        $(
-            #[cfg(feature = "time")]
-            #[cfg_attr(docsrs, doc(cfg(feature = "time")))]
-            $item
-        )*
-    }
-}
-
-macro_rules! cfg_sync {
-    ($($item:item)*) => {
-        $(
-            #[cfg(feature = "sync")]
-            #[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
-            $item
-        )*
-    }
-}
-
-macro_rules! cfg_signal {
-    ($($item:item)*) => {
-        $(
-            #[cfg(feature = "signal")]
-            #[cfg_attr(docsrs, doc(cfg(feature = "signal")))]
-            $item
-        )*
-    }
-}
-
-macro_rules! ready {
-    ($e:expr $(,)?) => {
-        match $e {
-            std::task::Poll::Ready(t) => t,
-            std::task::Poll::Pending => return std::task::Poll::Pending,
-        }
-    };
-}
@@ -1,52 +0,0 @@
-use crate::Stream;
-
-use core::fmt;
-use core::pin::Pin;
-use core::task::{Context, Poll};
-use pin_project_lite::pin_project;
-
-pin_project! {
-    /// Stream for the [`map_while`](super::StreamExt::map_while) method.
-    #[must_use = "streams do nothing unless polled"]
-    pub struct MapWhile<St, F> {
-        #[pin]
-        stream: St,
-        f: F,
-    }
-}
-
-impl<St, F> fmt::Debug for MapWhile<St, F>
-where
-    St: fmt::Debug,
-{
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("MapWhile")
-            .field("stream", &self.stream)
-            .finish()
-    }
-}
-
-impl<St, F> MapWhile<St, F> {
-    pub(super) fn new(stream: St, f: F) -> Self {
-        MapWhile { stream, f }
-    }
-}
-
-impl<St, F, T> Stream for MapWhile<St, F>
-where
-    St: Stream,
-    F: FnMut(St::Item) -> Option<T>,
-{
-    type Item = T;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
-        let me = self.project();
-        let f = me.f;
-        me.stream.poll_next(cx).map(|opt| opt.and_then(f))
-    }
-
-    fn size_hint(&self) -> (usize, Option<usize>) {
-        let (_, upper) = self.stream.size_hint();
-        (0, upper)
-    }
-}
@@ -1,83 +0,0 @@
-use crate::Stream;
-
-use core::fmt;
-use core::future::Future;
-use core::pin::Pin;
-use core::task::{Context, Poll};
-use pin_project_lite::pin_project;
-
-pin_project! {
-    /// Stream for the [`then`](super::StreamExt::then) method.
-    #[must_use = "streams do nothing unless polled"]
-    pub struct Then<St, Fut, F> {
-        #[pin]
-        stream: St,
-        #[pin]
-        future: Option<Fut>,
-        f: F,
-    }
-}
-
-impl<St, Fut, F> fmt::Debug for Then<St, Fut, F>
-where
-    St: fmt::Debug,
-{
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("Then")
-            .field("stream", &self.stream)
-            .finish()
-    }
-}
-
-impl<St, Fut, F> Then<St, Fut, F> {
-    pub(super) fn new(stream: St, f: F) -> Self {
-        Then {
-            stream,
-            future: None,
-            f,
-        }
-    }
-}
-
-impl<St, F, Fut> Stream for Then<St, Fut, F>
-where
-    St: Stream,
-    Fut: Future,
-    F: FnMut(St::Item) -> Fut,
-{
-    type Item = Fut::Output;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Fut::Output>> {
-        let mut me = self.project();
-
-        loop {
-            if let Some(future) = me.future.as_mut().as_pin_mut() {
-                match future.poll(cx) {
-                    Poll::Ready(item) => {
-                        me.future.set(None);
-                        return Poll::Ready(Some(item));
-                    }
-                    Poll::Pending => return Poll::Pending,
-                }
-            }
-
-            match me.stream.as_mut().poll_next(cx) {
-                Poll::Ready(Some(item)) => {
-                    me.future.set(Some((me.f)(item)));
-                }
-                Poll::Ready(None) => return Poll::Ready(None),
-                Poll::Pending => return Poll::Pending,
-            }
-        }
-    }
-
-    fn size_hint(&self) -> (usize, Option<usize>) {
-        let future_len = if self.future.is_some() { 1 } else { 0 };
-        let (lower, upper) = self.stream.size_hint();
-
-        let lower = lower.saturating_add(future_len);
-        let upper = upper.and_then(|upper| upper.checked_add(future_len));
-
-        (lower, upper)
-    }
-}
@@ -1,107 +0,0 @@
-use crate::stream_ext::Fuse;
-use crate::Stream;
-use tokio::time::{Instant, Sleep};
-
-use core::future::Future;
-use core::pin::Pin;
-use core::task::{Context, Poll};
-use pin_project_lite::pin_project;
-use std::fmt;
-use std::time::Duration;
-
-pin_project! {
-    /// Stream returned by the [`timeout`](super::StreamExt::timeout) method.
-    #[must_use = "streams do nothing unless polled"]
-    #[derive(Debug)]
-    pub struct Timeout<S> {
-        #[pin]
-        stream: Fuse<S>,
-        #[pin]
-        deadline: Sleep,
-        duration: Duration,
-        poll_deadline: bool,
-    }
-}
-
-/// Error returned by `Timeout`.
-#[derive(Debug, PartialEq)]
-pub struct Elapsed(());
-
-impl<S: Stream> Timeout<S> {
-    pub(super) fn new(stream: S, duration: Duration) -> Self {
-        let next = Instant::now() + duration;
-        let deadline = tokio::time::sleep_until(next);
-
-        Timeout {
-            stream: Fuse::new(stream),
-            deadline,
-            duration,
-            poll_deadline: true,
-        }
-    }
-}
-
-impl<S: Stream> Stream for Timeout<S> {
-    type Item = Result<S::Item, Elapsed>;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let me = self.project();
-
-        match me.stream.poll_next(cx) {
-            Poll::Ready(v) => {
-                if v.is_some() {
-                    let next = Instant::now() + *me.duration;
-                    me.deadline.reset(next);
-                    *me.poll_deadline = true;
-                }
-                return Poll::Ready(v.map(Ok));
-            }
-            Poll::Pending => {}
-        };
-
-        if *me.poll_deadline {
-            ready!(me.deadline.poll(cx));
-            *me.poll_deadline = false;
-            return Poll::Ready(Some(Err(Elapsed::new())));
-        }
-
-        Poll::Pending
-    }
-
-    fn size_hint(&self) -> (usize, Option<usize>) {
-        let (lower, upper) = self.stream.size_hint();
-
-        // The timeout stream may insert an error before and after each message
-        // from the underlying stream, but no more than one error between each
-        // message. Hence the upper bound is computed as 2x+1.
-
-        // Using a helper function to enable use of question mark operator.
-        fn twice_plus_one(value: Option<usize>) -> Option<usize> {
-            value?.checked_mul(2)?.checked_add(1)
-        }
-
-        (lower, twice_plus_one(upper))
-    }
-}
-
-// ===== impl Elapsed =====
-
-impl Elapsed {
-    pub(crate) fn new() -> Self {
-        Elapsed(())
-    }
-}
-
-impl fmt::Display for Elapsed {
-    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
-        "deadline has elapsed".fmt(fmt)
-    }
-}
-
-impl std::error::Error for Elapsed {}
-
-impl From<Elapsed> for std::io::Error {
-    fn from(_err: Elapsed) -> std::io::Error {
-        std::io::ErrorKind::TimedOut.into()
-    }
-}
@@ -1,62 +0,0 @@
-//! Wrappers for Tokio types that implement `Stream`.
-
-/// Error types for the wrappers.
-pub mod errors {
-    cfg_sync! {
-        pub use crate::wrappers::broadcast::BroadcastStreamRecvError;
-    }
-}
-
-mod mpsc_bounded;
-pub use mpsc_bounded::ReceiverStream;
-
-mod mpsc_unbounded;
-pub use mpsc_unbounded::UnboundedReceiverStream;
-
-cfg_sync! {
-    mod broadcast;
-    pub use broadcast::BroadcastStream;
-
-    mod watch;
-    pub use watch::WatchStream;
-}
-
-cfg_signal! {
-    #[cfg(unix)]
-    mod signal_unix;
-    #[cfg(unix)]
-    pub use signal_unix::SignalStream;
-
-    #[cfg(any(windows, docsrs))]
-    mod signal_windows;
-    #[cfg(any(windows, docsrs))]
-    pub use signal_windows::{CtrlCStream, CtrlBreakStream};
-}
-
-cfg_time! {
-    mod interval;
-    pub use interval::IntervalStream;
-}
-
-cfg_net! {
-    mod tcp_listener;
-    pub use tcp_listener::TcpListenerStream;
-
-    #[cfg(unix)]
-    mod unix_listener;
-    #[cfg(unix)]
-    pub use unix_listener::UnixListenerStream;
-}
-
-cfg_io_util! {
-    mod split;
-    pub use split::SplitStream;
-
-    mod lines;
-    pub use lines::LinesStream;
-}
-
-cfg_fs! {
-    mod read_dir;
-    pub use read_dir::ReadDirStream;
-}
@@ -1,79 +0,0 @@
-use std::pin::Pin;
-use tokio::sync::broadcast::error::RecvError;
-use tokio::sync::broadcast::Receiver;
-
-use futures_core::Stream;
-use tokio_util::sync::ReusableBoxFuture;
-
-use std::fmt;
-use std::task::{Context, Poll};
-
-/// A wrapper around [`tokio::sync::broadcast::Receiver`] that implements [`Stream`].
-///
-/// [`tokio::sync::broadcast::Receiver`]: struct@tokio::sync::broadcast::Receiver
-/// [`Stream`]: trait@crate::Stream
-#[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
-pub struct BroadcastStream<T> {
-    inner: ReusableBoxFuture<'static, (Result<T, RecvError>, Receiver<T>)>,
-}
-
-/// An error returned from the inner stream of a [`BroadcastStream`].
-#[derive(Debug, PartialEq, Clone)]
-pub enum BroadcastStreamRecvError {
-    /// The receiver lagged too far behind. Attempting to receive again will
-    /// return the oldest message still retained by the channel.
-    ///
-    /// Includes the number of skipped messages.
-    Lagged(u64),
-}
-
-impl fmt::Display for BroadcastStreamRecvError {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            BroadcastStreamRecvError::Lagged(amt) => write!(f, "channel lagged by {}", amt),
-        }
-    }
-}
-
-impl std::error::Error for BroadcastStreamRecvError {}
-
-async fn make_future<T: Clone>(mut rx: Receiver<T>) -> (Result<T, RecvError>, Receiver<T>) {
-    let result = rx.recv().await;
-    (result, rx)
-}
-
-impl<T: 'static + Clone + Send> BroadcastStream<T> {
-    /// Create a new `BroadcastStream`.
-    pub fn new(rx: Receiver<T>) -> Self {
-        Self {
-            inner: ReusableBoxFuture::new(make_future(rx)),
-        }
-    }
-}
-
-impl<T: 'static + Clone + Send> Stream for BroadcastStream<T> {
-    type Item = Result<T, BroadcastStreamRecvError>;
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let (result, rx) = ready!(self.inner.poll(cx));
-        self.inner.set(make_future(rx));
-        match result {
-            Ok(item) => Poll::Ready(Some(Ok(item))),
-            Err(RecvError::Closed) => Poll::Ready(None),
-            Err(RecvError::Lagged(n)) => {
-                Poll::Ready(Some(Err(BroadcastStreamRecvError::Lagged(n))))
-            }
-        }
-    }
-}
-
-impl<T> fmt::Debug for BroadcastStream<T> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("BroadcastStream").finish()
-    }
-}
-
-impl<T: 'static + Clone + Send> From<Receiver<T>> for BroadcastStream<T> {
-    fn from(recv: Receiver<T>) -> Self {
-        Self::new(recv)
-    }
-}
@@ -1,50 +0,0 @@
-use crate::Stream;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use tokio::time::{Instant, Interval};
-
-/// A wrapper around [`Interval`] that implements [`Stream`].
-///
-/// [`Interval`]: struct@tokio::time::Interval
-/// [`Stream`]: trait@crate::Stream
-#[derive(Debug)]
-#[cfg_attr(docsrs, doc(cfg(feature = "time")))]
-pub struct IntervalStream {
-    inner: Interval,
-}
-
-impl IntervalStream {
-    /// Create a new `IntervalStream`.
-    pub fn new(interval: Interval) -> Self {
-        Self { inner: interval }
-    }
-
-    /// Get back the inner `Interval`.
-    pub fn into_inner(self) -> Interval {
-        self.inner
-    }
-}
-
-impl Stream for IntervalStream {
-    type Item = Instant;
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Instant>> {
-        self.inner.poll_tick(cx).map(Some)
-    }
-
-    fn size_hint(&self) -> (usize, Option<usize>) {
-        (std::usize::MAX, None)
-    }
-}
-
-impl AsRef<Interval> for IntervalStream {
-    fn as_ref(&self) -> &Interval {
-        &self.inner
-    }
-}
-
-impl AsMut<Interval> for IntervalStream {
-    fn as_mut(&mut self) -> &mut Interval {
-        &mut self.inner
-    }
-}
@@ -1,60 +0,0 @@
-use crate::Stream;
-use pin_project_lite::pin_project;
-use std::io;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use tokio::io::{AsyncBufRead, Lines};
-
-pin_project! {
-    /// A wrapper around [`tokio::io::Lines`] that implements [`Stream`].
-    ///
-    /// [`tokio::io::Lines`]: struct@tokio::io::Lines
-    /// [`Stream`]: trait@crate::Stream
-    #[derive(Debug)]
-    #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
-    pub struct LinesStream<R> {
-        #[pin]
-        inner: Lines<R>,
-    }
-}
-
-impl<R> LinesStream<R> {
-    /// Create a new `LinesStream`.
-    pub fn new(lines: Lines<R>) -> Self {
-        Self { inner: lines }
-    }
-
-    /// Get back the inner `Lines`.
-    pub fn into_inner(self) -> Lines<R> {
-        self.inner
-    }
-
-    /// Obtain a pinned reference to the inner `Lines<R>`.
-    #[allow(clippy::wrong_self_convention)] // https://github.com/rust-lang/rust-clippy/issues/4546
-    pub fn as_pin_mut(self: Pin<&mut Self>) -> Pin<&mut Lines<R>> {
-        self.project().inner
-    }
-}
-
-impl<R: AsyncBufRead> Stream for LinesStream<R> {
-    type Item = io::Result<String>;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.project()
-            .inner
-            .poll_next_line(cx)
-            .map(Result::transpose)
-    }
-}
-
-impl<R> AsRef<Lines<R>> for LinesStream<R> {
-    fn as_ref(&self) -> &Lines<R> {
-        &self.inner
-    }
-}
-
-impl<R> AsMut<Lines<R>> for LinesStream<R> {
-    fn as_mut(&mut self) -> &mut Lines<R> {
-        &mut self.inner
-    }
-}
@@ -1,65 +0,0 @@
-use crate::Stream;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use tokio::sync::mpsc::Receiver;
-
-/// A wrapper around [`tokio::sync::mpsc::Receiver`] that implements [`Stream`].
-///
-/// [`tokio::sync::mpsc::Receiver`]: struct@tokio::sync::mpsc::Receiver
-/// [`Stream`]: trait@crate::Stream
-#[derive(Debug)]
-pub struct ReceiverStream<T> {
-    inner: Receiver<T>,
-}
-
-impl<T> ReceiverStream<T> {
-    /// Create a new `ReceiverStream`.
-    pub fn new(recv: Receiver<T>) -> Self {
-        Self { inner: recv }
-    }
-
-    /// Get back the inner `Receiver`.
-    pub fn into_inner(self) -> Receiver<T> {
-        self.inner
-    }
-
-    /// Closes the receiving half of a channel without dropping it.
-    ///
-    /// This prevents any further messages from being sent on the channel while
-    /// still enabling the receiver to drain messages that are buffered. Any
-    /// outstanding [`Permit`] values will still be able to send messages.
-    ///
-    /// To guarantee no messages are dropped, after calling `close()`, you must
-    /// receive all items from the stream until `None` is returned.
-    ///
-    /// [`Permit`]: struct@tokio::sync::mpsc::Permit
-    pub fn close(&mut self) {
-        self.inner.close()
-    }
-}
-
-impl<T> Stream for ReceiverStream<T> {
-    type Item = T;
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.inner.poll_recv(cx)
-    }
-}
-
-impl<T> AsRef<Receiver<T>> for ReceiverStream<T> {
-    fn as_ref(&self) -> &Receiver<T> {
-        &self.inner
-    }
-}
-
-impl<T> AsMut<Receiver<T>> for ReceiverStream<T> {
-    fn as_mut(&mut self) -> &mut Receiver<T> {
-        &mut self.inner
-    }
-}
-
-impl<T> From<Receiver<T>> for ReceiverStream<T> {
-    fn from(recv: Receiver<T>) -> Self {
-        Self::new(recv)
-    }
-}
@@ -1,59 +0,0 @@
-use crate::Stream;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use tokio::sync::mpsc::UnboundedReceiver;
-
-/// A wrapper around [`tokio::sync::mpsc::UnboundedReceiver`] that implements [`Stream`].
-///
-/// [`tokio::sync::mpsc::UnboundedReceiver`]: struct@tokio::sync::mpsc::UnboundedReceiver
-/// [`Stream`]: trait@crate::Stream
-#[derive(Debug)]
-pub struct UnboundedReceiverStream<T> {
-    inner: UnboundedReceiver<T>,
-}
-
-impl<T> UnboundedReceiverStream<T> {
-    /// Create a new `UnboundedReceiverStream`.
-    pub fn new(recv: UnboundedReceiver<T>) -> Self {
-        Self { inner: recv }
-    }
-
-    /// Get back the inner `UnboundedReceiver`.
-    pub fn into_inner(self) -> UnboundedReceiver<T> {
-        self.inner
-    }
-
-    /// Closes the receiving half of a channel without dropping it.
-    ///
-    /// This prevents any further messages from being sent on the channel while
-    /// still enabling the receiver to drain messages that are buffered.
-    pub fn close(&mut self) {
-        self.inner.close()
-    }
-}
-
-impl<T> Stream for UnboundedReceiverStream<T> {
-    type Item = T;
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.inner.poll_recv(cx)
-    }
-}
-
-impl<T> AsRef<UnboundedReceiver<T>> for UnboundedReceiverStream<T> {
-    fn as_ref(&self) -> &UnboundedReceiver<T> {
-        &self.inner
-    }
-}
-
-impl<T> AsMut<UnboundedReceiver<T>> for UnboundedReceiverStream<T> {
-    fn as_mut(&mut self) -> &mut UnboundedReceiver<T> {
-        &mut self.inner
-    }
-}
-
-impl<T> From<UnboundedReceiver<T>> for UnboundedReceiverStream<T> {
-    fn from(recv: UnboundedReceiver<T>) -> Self {
-        Self::new(recv)
-    }
-}
@@ -1,47 +0,0 @@
-use crate::Stream;
-use std::io;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use tokio::fs::{DirEntry, ReadDir};
-
-/// A wrapper around [`tokio::fs::ReadDir`] that implements [`Stream`].
-///
-/// [`tokio::fs::ReadDir`]: struct@tokio::fs::ReadDir
-/// [`Stream`]: trait@crate::Stream
-#[derive(Debug)]
-#[cfg_attr(docsrs, doc(cfg(feature = "fs")))]
-pub struct ReadDirStream {
-    inner: ReadDir,
-}
-
-impl ReadDirStream {
-    /// Create a new `ReadDirStream`.
-    pub fn new(read_dir: ReadDir) -> Self {
-        Self { inner: read_dir }
-    }
-
-    /// Get back the inner `ReadDir`.
-    pub fn into_inner(self) -> ReadDir {
-        self.inner
-    }
-}
-
-impl Stream for ReadDirStream {
-    type Item = io::Result<DirEntry>;
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.inner.poll_next_entry(cx).map(Result::transpose)
-    }
-}
-
-impl AsRef<ReadDir> for ReadDirStream {
-    fn as_ref(&self) -> &ReadDir {
-        &self.inner
-    }
-}
-
-impl AsMut<ReadDir> for ReadDirStream {
-    fn as_mut(&mut self) -> &mut ReadDir {
-        &mut self.inner
-    }
-}
@@ -1,46 +0,0 @@
-use crate::Stream;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use tokio::signal::unix::Signal;
-
-/// A wrapper around [`Signal`] that implements [`Stream`].
-///
-/// [`Signal`]: struct@tokio::signal::unix::Signal
-/// [`Stream`]: trait@crate::Stream
-#[derive(Debug)]
-#[cfg_attr(docsrs, doc(cfg(all(unix, feature = "signal"))))]
-pub struct SignalStream {
-    inner: Signal,
-}
-
-impl SignalStream {
-    /// Create a new `SignalStream`.
-    pub fn new(interval: Signal) -> Self {
-        Self { inner: interval }
-    }
-
-    /// Get back the inner `Signal`.
-    pub fn into_inner(self) -> Signal {
-        self.inner
-    }
-}
-
-impl Stream for SignalStream {
-    type Item = ();
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<()>> {
-        self.inner.poll_recv(cx)
-    }
-}
-
-impl AsRef<Signal> for SignalStream {
-    fn as_ref(&self) -> &Signal {
-        &self.inner
-    }
-}
-
-impl AsMut<Signal> for SignalStream {
-    fn as_mut(&mut self) -> &mut Signal {
-        &mut self.inner
-    }
-}
@@ -1,88 +0,0 @@
-use crate::Stream;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use tokio::signal::windows::{CtrlBreak, CtrlC};
-
-/// A wrapper around [`CtrlC`] that implements [`Stream`].
-///
-/// [`CtrlC`]: struct@tokio::signal::windows::CtrlC
-/// [`Stream`]: trait@crate::Stream
-#[derive(Debug)]
-#[cfg_attr(docsrs, doc(cfg(all(windows, feature = "signal"))))]
-pub struct CtrlCStream {
-    inner: CtrlC,
-}
-
-impl CtrlCStream {
-    /// Create a new `CtrlCStream`.
-    pub fn new(interval: CtrlC) -> Self {
-        Self { inner: interval }
-    }
-
-    /// Get back the inner `CtrlC`.
-    pub fn into_inner(self) -> CtrlC {
-        self.inner
-    }
-}
-
-impl Stream for CtrlCStream {
-    type Item = ();
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<()>> {
-        self.inner.poll_recv(cx)
-    }
-}
-
-impl AsRef<CtrlC> for CtrlCStream {
-    fn as_ref(&self) -> &CtrlC {
-        &self.inner
-    }
-}
-
-impl AsMut<CtrlC> for CtrlCStream {
-    fn as_mut(&mut self) -> &mut CtrlC {
-        &mut self.inner
-    }
-}
-
-/// A wrapper around [`CtrlBreak`] that implements [`Stream`].
-///
-/// [`CtrlBreak`]: struct@tokio::signal::windows::CtrlBreak
-/// [`Stream`]: trait@crate::Stream
-#[derive(Debug)]
-#[cfg_attr(docsrs, doc(cfg(all(windows, feature = "signal"))))]
-pub struct CtrlBreakStream {
-    inner: CtrlBreak,
-}
-
-impl CtrlBreakStream {
-    /// Create a new `CtrlBreakStream`.
-    pub fn new(interval: CtrlBreak) -> Self {
-        Self { inner: interval }
-    }
-
-    /// Get back the inner `CtrlBreak`.
-    pub fn into_inner(self) -> CtrlBreak {
-        self.inner
-    }
-}
-
-impl Stream for CtrlBreakStream {
-    type Item = ();
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<()>> {
-        self.inner.poll_recv(cx)
-    }
-}
-
-impl AsRef<CtrlBreak> for CtrlBreakStream {
-    fn as_ref(&self) -> &CtrlBreak {
-        &self.inner
-    }
-}
-
-impl AsMut<CtrlBreak> for CtrlBreakStream {
-    fn as_mut(&mut self) -> &mut CtrlBreak {
-        &mut self.inner
-    }
-}
@@ -1,60 +0,0 @@
-use crate::Stream;
-use pin_project_lite::pin_project;
-use std::io;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use tokio::io::{AsyncBufRead, Split};
-
-pin_project! {
-    /// A wrapper around [`tokio::io::Split`] that implements [`Stream`].
-    ///
-    /// [`tokio::io::Split`]: struct@tokio::io::Split
-    /// [`Stream`]: trait@crate::Stream
-    #[derive(Debug)]
-    #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
-    pub struct SplitStream<R> {
-        #[pin]
-        inner: Split<R>,
-    }
-}
-
-impl<R> SplitStream<R> {
-    /// Create a new `SplitStream`.
-    pub fn new(split: Split<R>) -> Self {
-        Self { inner: split }
-    }
-
-    /// Get back the inner `Split`.
-    pub fn into_inner(self) -> Split<R> {
-        self.inner
-    }
-
-    /// Obtain a pinned reference to the inner `Split<R>`.
-    #[allow(clippy::wrong_self_convention)] // https://github.com/rust-lang/rust-clippy/issues/4546
-    pub fn as_pin_mut(self: Pin<&mut Self>) -> Pin<&mut Split<R>> {
-        self.project().inner
-    }
-}
-
-impl<R: AsyncBufRead> Stream for SplitStream<R> {
-    type Item = io::Result<Vec<u8>>;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        self.project()
-            .inner
-            .poll_next_segment(cx)
-            .map(Result::transpose)
-    }
-}
-
-impl<R> AsRef<Split<R>> for SplitStream<R> {
-    fn as_ref(&self) -> &Split<R> {
-        &self.inner
-    }
-}
-
-impl<R> AsMut<Split<R>> for SplitStream<R> {
-    fn as_mut(&mut self) -> &mut Split<R> {
-        &mut self.inner
-    }
-}
@@ -1,54 +0,0 @@
-use crate::Stream;
-use std::io;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use tokio::net::{TcpListener, TcpStream};
-
-/// A wrapper around [`TcpListener`] that implements [`Stream`].
-///
-/// [`TcpListener`]: struct@tokio::net::TcpListener
-/// [`Stream`]: trait@crate::Stream
-#[derive(Debug)]
-#[cfg_attr(docsrs, doc(cfg(feature = "net")))]
-pub struct TcpListenerStream {
-    inner: TcpListener,
-}
-
-impl TcpListenerStream {
-    /// Create a new `TcpListenerStream`.
-    pub fn new(listener: TcpListener) -> Self {
-        Self { inner: listener }
-    }
-
-    /// Get back the inner `TcpListener`.
-    pub fn into_inner(self) -> TcpListener {
-        self.inner
-    }
-}
-
-impl Stream for TcpListenerStream {
-    type Item = io::Result<TcpStream>;
-
-    fn poll_next(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Option<io::Result<TcpStream>>> {
-        match self.inner.poll_accept(cx) {
-            Poll::Ready(Ok((stream, _))) => Poll::Ready(Some(Ok(stream))),
-            Poll::Ready(Err(err)) => Poll::Ready(Some(Err(err))),
-            Poll::Pending => Poll::Pending,
-        }
-    }
-}
-
-impl AsRef<TcpListener> for TcpListenerStream {
-    fn as_ref(&self) -> &TcpListener {
-        &self.inner
-    }
-}
-
-impl AsMut<TcpListener> for TcpListenerStream {
-    fn as_mut(&mut self) -> &mut TcpListener {
-        &mut self.inner
-    }
-}
@@ -1,54 +0,0 @@
-use crate::Stream;
-use std::io;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use tokio::net::{UnixListener, UnixStream};
-
-/// A wrapper around [`UnixListener`] that implements [`Stream`].
-///
-/// [`UnixListener`]: struct@tokio::net::UnixListener
-/// [`Stream`]: trait@crate::Stream
-#[derive(Debug)]
-#[cfg_attr(docsrs, doc(cfg(all(unix, feature = "net"))))]
-pub struct UnixListenerStream {
-    inner: UnixListener,
-}
-
-impl UnixListenerStream {
-    /// Create a new `UnixListenerStream`.
-    pub fn new(listener: UnixListener) -> Self {
-        Self { inner: listener }
-    }
-
-    /// Get back the inner `UnixListener`.
-    pub fn into_inner(self) -> UnixListener {
-        self.inner
-    }
-}
-
-impl Stream for UnixListenerStream {
-    type Item = io::Result<UnixStream>;
-
-    fn poll_next(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Option<io::Result<UnixStream>>> {
-        match self.inner.poll_accept(cx) {
-            Poll::Ready(Ok((stream, _))) => Poll::Ready(Some(Ok(stream))),
-            Poll::Ready(Err(err)) => Poll::Ready(Some(Err(err))),
-            Poll::Pending => Poll::Pending,
-        }
-    }
-}
-
-impl AsRef<UnixListener> for UnixListenerStream {
-    fn as_ref(&self) -> &UnixListener {
-        &self.inner
-    }
-}
-
-impl AsMut<UnixListener> for UnixListenerStream {
-    fn as_mut(&mut self) -> &mut UnixListener {
-        &mut self.inner
-    }
-}
@@ -1,102 +0,0 @@
-use std::pin::Pin;
-use tokio::sync::watch::Receiver;
-
-use futures_core::Stream;
-use tokio_util::sync::ReusableBoxFuture;
-
-use std::fmt;
-use std::task::{Context, Poll};
-use tokio::sync::watch::error::RecvError;
-
-/// A wrapper around [`tokio::sync::watch::Receiver`] that implements [`Stream`].
-///
-/// This stream will always start by yielding the current value when the WatchStream is polled,
-/// regardless of whether it was the initial value or sent afterwards.
-///
-/// # Examples
-///
-/// ```
-/// # #[tokio::main]
-/// # async fn main() {
-/// use tokio_stream::{StreamExt, wrappers::WatchStream};
-/// use tokio::sync::watch;
-///
-/// let (tx, rx) = watch::channel("hello");
-/// let mut rx = WatchStream::new(rx);
-///
-/// assert_eq!(rx.next().await, Some("hello"));
-///
-/// tx.send("goodbye").unwrap();
-/// assert_eq!(rx.next().await, Some("goodbye"));
-/// # }
-/// ```
-///
-/// ```
-/// # #[tokio::main]
-/// # async fn main() {
-/// use tokio_stream::{StreamExt, wrappers::WatchStream};
-/// use tokio::sync::watch;
-///
-/// let (tx, rx) = watch::channel("hello");
-/// let mut rx = WatchStream::new(rx);
-///
-/// tx.send("goodbye").unwrap();
-/// assert_eq!(rx.next().await, Some("goodbye"));
-/// # }
-/// ```
-///
-/// [`tokio::sync::watch::Receiver`]: struct@tokio::sync::watch::Receiver
-/// [`Stream`]: trait@crate::Stream
-#[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
-pub struct WatchStream<T> {
-    inner: ReusableBoxFuture<'static, (Result<(), RecvError>, Receiver<T>)>,
-}
-
-async fn make_future<T: Clone + Send + Sync>(
-    mut rx: Receiver<T>,
-) -> (Result<(), RecvError>, Receiver<T>) {
-    let result = rx.changed().await;
-    (result, rx)
-}
-
-impl<T: 'static + Clone + Send + Sync> WatchStream<T> {
-    /// Create a new `WatchStream`.
-    pub fn new(rx: Receiver<T>) -> Self {
-        Self {
-            inner: ReusableBoxFuture::new(async move { (Ok(()), rx) }),
-        }
-    }
-}
-
-impl<T: Clone + 'static + Send + Sync> Stream for WatchStream<T> {
-    type Item = T;
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let (result, mut rx) = ready!(self.inner.poll(cx));
-        match result {
-            Ok(_) => {
-                let received = (*rx.borrow_and_update()).clone();
-                self.inner.set(make_future(rx));
-                Poll::Ready(Some(received))
-            }
-            Err(_) => {
-                self.inner.set(make_future(rx));
-                Poll::Ready(None)
-            }
-        }
-    }
-}
-
-impl<T> Unpin for WatchStream<T> {}
-
-impl<T> fmt::Debug for WatchStream<T> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("WatchStream").finish()
-    }
-}
-
-impl<T: 'static + Clone + Send + Sync> From<Receiver<T>> for WatchStream<T> {
-    fn from(recv: Receiver<T>) -> Self {
-        Self::new(recv)
-    }
-}
@@ -1,107 +0,0 @@
-#![allow(clippy::diverging_sub_expression)]
-
-use std::rc::Rc;
-
-#[allow(dead_code)]
-type BoxStream<T> = std::pin::Pin<Box<dyn tokio_stream::Stream<Item = T>>>;
-
-#[allow(dead_code)]
-fn require_send<T: Send>(_t: &T) {}
-#[allow(dead_code)]
-fn require_sync<T: Sync>(_t: &T) {}
-#[allow(dead_code)]
-fn require_unpin<T: Unpin>(_t: &T) {}
-
-#[allow(dead_code)]
-struct Invalid;
-
-trait AmbiguousIfSend<A> {
-    fn some_item(&self) {}
-}
-impl<T: ?Sized> AmbiguousIfSend<()> for T {}
-impl<T: ?Sized + Send> AmbiguousIfSend<Invalid> for T {}
-
-trait AmbiguousIfSync<A> {
-    fn some_item(&self) {}
-}
-impl<T: ?Sized> AmbiguousIfSync<()> for T {}
-impl<T: ?Sized + Sync> AmbiguousIfSync<Invalid> for T {}
-
-trait AmbiguousIfUnpin<A> {
-    fn some_item(&self) {}
-}
-impl<T: ?Sized> AmbiguousIfUnpin<()> for T {}
-impl<T: ?Sized + Unpin> AmbiguousIfUnpin<Invalid> for T {}
-
-macro_rules! into_todo {
-    ($typ:ty) => {{
-        let x: $typ = todo!();
-        x
-    }};
-}
-
-macro_rules! async_assert_fn {
-    ($($f:ident $(< $($generic:ty),* > )? )::+($($arg:ty),*): Send & Sync) => {
-        #[allow(unreachable_code)]
-        #[allow(unused_variables)]
-        const _: fn() = || {
-            let f = $($f $(::<$($generic),*>)? )::+( $( into_todo!($arg) ),* );
-            require_send(&f);
-            require_sync(&f);
-        };
-    };
-    ($($f:ident $(< $($generic:ty),* > )? )::+($($arg:ty),*): Send & !Sync) => {
-        #[allow(unreachable_code)]
-        #[allow(unused_variables)]
-        const _: fn() = || {
-            let f = $($f $(::<$($generic),*>)? )::+( $( into_todo!($arg) ),* );
-            require_send(&f);
-            AmbiguousIfSync::some_item(&f);
-        };
-    };
-    ($($f:ident $(< $($generic:ty),* > )? )::+($($arg:ty),*): !Send & Sync) => {
-        #[allow(unreachable_code)]
-        #[allow(unused_variables)]
-        const _: fn() = || {
-            let f = $($f $(::<$($generic),*>)? )::+( $( into_todo!($arg) ),* );
-            AmbiguousIfSend::some_item(&f);
-            require_sync(&f);
-        };
-    };
-    ($($f:ident $(< $($generic:ty),* > )? )::+($($arg:ty),*): !Send & !Sync) => {
-        #[allow(unreachable_code)]
-        #[allow(unused_variables)]
-        const _: fn() = || {
-            let f = $($f $(::<$($generic),*>)? )::+( $( into_todo!($arg) ),* );
-            AmbiguousIfSend::some_item(&f);
-            AmbiguousIfSync::some_item(&f);
-        };
-    };
-    ($($f:ident $(< $($generic:ty),* > )? )::+($($arg:ty),*): !Unpin) => {
-        #[allow(unreachable_code)]
-        #[allow(unused_variables)]
-        const _: fn() = || {
-            let f = $($f $(::<$($generic),*>)? )::+( $( into_todo!($arg) ),* );
-            AmbiguousIfUnpin::some_item(&f);
-        };
-    };
-    ($($f:ident $(< $($generic:ty),* > )? )::+($($arg:ty),*): Unpin) => {
-        #[allow(unreachable_code)]
-        #[allow(unused_variables)]
-        const _: fn() = || {
-            let f = $($f $(::<$($generic),*>)? )::+( $( into_todo!($arg) ),* );
-            require_unpin(&f);
-        };
-    };
-}
-
-async_assert_fn!(tokio_stream::empty<Rc<u8>>(): Send & Sync);
-async_assert_fn!(tokio_stream::pending<Rc<u8>>(): Send & Sync);
-async_assert_fn!(tokio_stream::iter(std::vec::IntoIter<u8>): Send & Sync);
-
-async_assert_fn!(tokio_stream::StreamExt::next(&mut BoxStream<()>): !Unpin);
-async_assert_fn!(tokio_stream::StreamExt::try_next(&mut BoxStream<Result<(), ()>>): !Unpin);
-async_assert_fn!(tokio_stream::StreamExt::all(&mut BoxStream<()>, fn(())->bool): !Unpin);
-async_assert_fn!(tokio_stream::StreamExt::any(&mut BoxStream<()>, fn(())->bool): !Unpin);
-async_assert_fn!(tokio_stream::StreamExt::fold(&mut BoxStream<()>, (), fn((), ())->()): !Unpin);
-async_assert_fn!(tokio_stream::StreamExt::collect<Vec<()>>(&mut BoxStream<()>): !Unpin);
@@ -1,15 +0,0 @@
-use async_stream::stream;
-use tokio::sync::mpsc::{self, UnboundedSender};
-use tokio_stream::Stream;
-
-pub fn unbounded_channel_stream<T: Unpin>() -> (UnboundedSender<T>, impl Stream<Item = T>) {
-    let (tx, mut rx) = mpsc::unbounded_channel();
-
-    let stream = stream! {
-        while let Some(item) = rx.recv().await {
-            yield item;
-        }
-    };
-
-    (tx, stream)
-}
@@ -1,29 +0,0 @@
-#![cfg(feature = "sync")]
-
-use tokio::sync::watch;
-use tokio_stream::wrappers::WatchStream;
-use tokio_stream::StreamExt;
-
-#[tokio::test]
-async fn message_not_twice() {
-    let (tx, rx) = watch::channel("hello");
-
-    let mut counter = 0;
-    let mut stream = WatchStream::new(rx).map(move |payload| {
-        println!("{}", payload);
-        if payload == "goodbye" {
-            counter += 1;
-        }
-        if counter >= 2 {
-            panic!("too many goodbyes");
-        }
-    });
-
-    let task = tokio::spawn(async move { while stream.next().await.is_some() {} });
-
-    // Send goodbye just once
-    tx.send("goodbye").unwrap();
-
-    drop(tx);
-    task.await.unwrap();
-}
@@ -1,19 +1,3 @@
-# 0.4.2 (May 14, 2021)
-
- test: add `assert_elapsed!` macro ([#3728])
-
-[#3728]: https://github.com/tokio-rs/tokio/pull/3728
-
-# 0.4.1 (March 10, 2021)
-
- Fix `io::Mock` to be `Send` and `Sync` ([#3594])
-
-[#3594]: https://github.com/tokio-rs/tokio/pull/3594
-
-# 0.4.0 (December 23, 2020)
-
- Track `tokio` 1.0 release.
-
 # 0.3.0 (October 15, 2020)

 - Track `tokio` 0.3 release.
@@ -2,30 +2,31 @@
 name = "tokio-test"
 # When releasing to crates.io:
 # - Remove path dependencies
+# - Update html_root_url.
+# - Update doc url
+#   - Cargo.toml
 # - Update CHANGELOG.md.
-# - Create "tokio-test-0.4.x" git tag.
-version = "0.4.2"
+# - Create "v0.3.x" git tag.
+version = "0.3.0"
 edition = "2018"
-rust-version = "1.49"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
+documentation = "https://docs.rs/tokio-test/0.3.0/tokio_test"
 description = """
 Testing utilities for Tokio- and futures-based code
 """
 categories = ["asynchronous", "testing"]

 [dependencies]
-tokio = { version = "1.2.0", path = "../tokio", features = ["rt", "sync", "time", "test-util"] }
-tokio-stream = { version = "0.1.1", path = "../tokio-stream" }
-async-stream = "0.3"
+tokio = { version = "0.3.0", path = "../tokio", features = ["rt", "stream", "sync", "time", "test-util"] }

-bytes = "1.0.0"
+bytes = "0.6.0"
 futures-core = "0.3.0"

 [dev-dependencies]
-tokio = { version = "1.2.0", path = "../tokio", features = ["full"] }
+tokio = { version = "0.3.0", path = "../tokio", features = ["full"] }
 futures-util = "0.3.0"

 [package.metadata.docs.rs]
@@ -1,4 +1,4 @@
-Copyright (c) 2022 Tokio Contributors
+Copyright (c) 2019 Tokio Contributors

 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated
@@ -21,11 +21,9 @@
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::sync::mpsc;
 use tokio::time::{self, Duration, Instant, Sleep};
-use tokio_stream::wrappers::UnboundedReceiverStream;

-use futures_core::{ready, Stream};
+use futures_core::ready;
 use std::collections::VecDeque;
-use std::fmt;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -65,12 +63,13 @@ enum Action {
    WriteError(Option<Arc<io::Error>>),
 }

+#[derive(Debug)]
 struct Inner {
    actions: VecDeque<Action>,
    waiting: Option<Instant>,
-    sleep: Option<Pin<Box<Sleep>>>,
+    sleep: Option<Sleep>,
    read_wait: Option<Waker>,
-    rx: UnboundedReceiverStream<Action>,
+    rx: mpsc::UnboundedReceiver<Action>,
 }

 impl Builder {
@@ -187,8 +186,6 @@ impl Inner {
    fn new(actions: VecDeque<Action>) -> (Inner, Handle) {
        let (tx, rx) = mpsc::unbounded_channel();

-        let rx = UnboundedReceiverStream::new(rx);
-
        let inner = Inner {
            actions,
            sleep: None,
@@ -203,6 +200,8 @@ impl Inner {
    }

    fn poll_action(&mut self, cx: &mut task::Context<'_>) -> Poll<Option<Action>> {
+        use futures_core::stream::Stream;
+
        Pin::new(&mut self.rx).poll_next(cx)
    }

@@ -366,7 +365,7 @@ impl AsyncRead for Mock {
                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                    if let Some(rem) = self.inner.remaining_wait() {
                        let until = Instant::now() + rem;
-                        self.inner.sleep = Some(Box::pin(time::sleep_until(until)));
+                        self.inner.sleep = Some(time::sleep_until(until));
                    } else {
                        self.inner.read_wait = Some(cx.waker().clone());
                        return Poll::Pending;
@@ -411,7 +410,7 @@ impl AsyncWrite for Mock {
                Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                    if let Some(rem) = self.inner.remaining_wait() {
                        let until = Instant::now() + rem;
-                        self.inner.sleep = Some(Box::pin(time::sleep_until(until)));
+                        self.inner.sleep = Some(time::sleep_until(until));
                    } else {
                        panic!("unexpected WouldBlock");
                    }
@@ -486,9 +485,3 @@ fn is_task_ctx() -> bool {
    r
 }
 */
-
-impl fmt::Debug for Inner {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "Inner {{...}}")
-    }
-}
@@ -1,15 +1,17 @@
+#![doc(html_root_url = "https://docs.rs/tokio-test/0.3.0")]
 #![warn(
    missing_debug_implementations,
    missing_docs,
    rust_2018_idioms,
    unreachable_pub
 )]
+#![cfg_attr(docsrs, deny(broken_intra_doc_links))]
 #![doc(test(
    no_crate_inject,
    attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))
 ))]

-//! Tokio and Futures based testing utilities
+//! Tokio and Futures based testing utilites

 pub mod io;

@@ -20,9 +22,9 @@ pub mod task;
 /// future completes.
 ///
 /// For more information, see the documentation for
-/// [`tokio::runtime::Runtime::block_on`][runtime-block-on].
+/// [`tokio::runtime::current_thread::Runtime::block_on`][runtime-block-on].
 ///
-/// [runtime-block-on]: https://docs.rs/tokio/1.3.0/tokio/runtime/struct.Runtime.html#method.block_on
+/// [runtime-block-on]: https://docs.rs/tokio/0.2.0-alpha.2/tokio/runtime/current_thread/struct.Runtime.html#method.block_on
 pub fn block_on<F: std::future::Future>(future: F) -> F::Output {
    use tokio::runtime;

@@ -259,37 +259,3 @@ macro_rules! assert_err {
        }
    }};
 }
-
-/// Asserts that an exact duration has elapsed since since the start instant ±1ms.
-///
-/// ```rust
-/// use tokio::time::{self, Instant};
-/// use std::time::Duration;
-/// use tokio_test::assert_elapsed;
-/// # async fn test_time_passed() {
-///
-/// let start = Instant::now();
-/// let dur = Duration::from_millis(50);
-/// time::sleep(dur).await;
-/// assert_elapsed!(start, dur);
-/// # }
-/// ```
-///
-/// This 1ms buffer is required because Tokio's hashed-wheel timer has finite time resolution and
-/// will not always sleep for the exact interval.
-#[macro_export]
-macro_rules! assert_elapsed {
-    ($start:expr, $dur:expr) => {{
-        let elapsed = $start.elapsed();
-        // type ascription improves compiler error when wrong type is passed
-        let lower: std::time::Duration = $dur;
-
-        // Handles ms rounding
-        assert!(
-            elapsed >= lower && elapsed <= lower + std::time::Duration::from_millis(1),
-            "actual = {:?}, expected = {:?}",
-            elapsed,
-            lower
-        );
-    }};
-}
@@ -9,7 +9,7 @@ use std::pin::Pin;
 use std::sync::{Arc, Condvar, Mutex};
 use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};

-use tokio_stream::Stream;
+use tokio::stream::Stream;

 /// TODO: dox
 pub fn spawn<T>(task: T) -> Spawn<T> {
@@ -180,7 +180,7 @@ impl ThreadWaker {
        }
    }

-    /// Clears any previously received wakes, avoiding potential spurious
+    /// Clears any previously received wakes, avoiding potential spurrious
    /// wake notifications. This should only be called immediately before running the
    /// task.
    fn clear(&self) {
@@ -1,179 +1,3 @@
-# 0.7.1 (February 21, 2022)
-
-### Added
-
- codec: add `length_field_type` to `LengthDelimitedCodec` builder ([#4508])
- io: add `StreamReader::into_inner_with_chunk()` ([#4559])
-
-### Changed
-
- switch from log to tracing ([#4539])
-
-### Fixed
-
- sync: fix waker update condition in `CancellationToken` ([#4497])
- bumped tokio dependency to 1.6 to satisfy minimum requirements ([#4490])
-
-[#4490]: https://github.com/tokio-rs/tokio/pull/4490
-[#4497]: https://github.com/tokio-rs/tokio/pull/4497
-[#4508]: https://github.com/tokio-rs/tokio/pull/4508
-[#4539]: https://github.com/tokio-rs/tokio/pull/4539
-[#4559]: https://github.com/tokio-rs/tokio/pull/4559
-
-# 0.7.0 (February 9, 2022)
-
-### Added
-
- task: add `spawn_pinned` ([#3370])
- time: add `shrink_to_fit` and `compact` methods to `DelayQueue` ([#4170])
- codec: improve `Builder::max_frame_length` docs ([#4352])
- codec: add mutable reference getters for codecs to pinned `Framed` ([#4372])
- net: add generic trait to combine `UnixListener` and `TcpListener` ([#4385])
- codec: implement `Framed::map_codec` ([#4427])
- codec: implement `Encoder<BytesMut>` for `BytesCodec` ([#4465])
-
-### Changed
-
- sync: add lifetime parameter to `ReusableBoxFuture` ([#3762])
- sync: refactored `PollSender<T>` to fix a subtly broken `Sink<T>` implementation ([#4214])
- time: remove error case from the infallible `DelayQueue::poll_elapsed` ([#4241])
-
-[#3370]: https://github.com/tokio-rs/tokio/pull/3370
-[#4170]: https://github.com/tokio-rs/tokio/pull/4170
-[#4352]: https://github.com/tokio-rs/tokio/pull/4352
-[#4372]: https://github.com/tokio-rs/tokio/pull/4372
-[#4385]: https://github.com/tokio-rs/tokio/pull/4385
-[#4427]: https://github.com/tokio-rs/tokio/pull/4427
-[#4465]: https://github.com/tokio-rs/tokio/pull/4465
-[#3762]: https://github.com/tokio-rs/tokio/pull/3762
-[#4214]: https://github.com/tokio-rs/tokio/pull/4214
-[#4241]: https://github.com/tokio-rs/tokio/pull/4241
-
-# 0.6.9 (October 29, 2021)
-
-### Added
-
- codec: implement `Clone` for `LengthDelimitedCodec` ([#4089])
- io: add `SyncIoBridge`  ([#4146])
-
-### Fixed
-
- time: update deadline on removal in `DelayQueue` ([#4178])
- codec: Update stream impl for Framed to return None after Err ([#4166])
-
-[#4089]: https://github.com/tokio-rs/tokio/pull/4089
-[#4146]: https://github.com/tokio-rs/tokio/pull/4146
-[#4166]: https://github.com/tokio-rs/tokio/pull/4166
-[#4178]: https://github.com/tokio-rs/tokio/pull/4178
-
-# 0.6.8 (September 3, 2021)
-
-### Added
-
- sync: add drop guard for `CancellationToken` ([#3839])
- compact: added `AsyncSeek` compat ([#4078])
- time: expose `Key` used in `DelayQueue`'s `Expired` ([#4081])
- io: add `with_capacity` to `ReaderStream` ([#4086])
-
-### Fixed
-
- codec: remove unnecessary `doc(cfg(...))` ([#3989])
-
-[#3839]: https://github.com/tokio-rs/tokio/pull/3839
-[#4078]: https://github.com/tokio-rs/tokio/pull/4078
-[#4081]: https://github.com/tokio-rs/tokio/pull/4081
-[#4086]: https://github.com/tokio-rs/tokio/pull/4086
-[#3989]: https://github.com/tokio-rs/tokio/pull/3989
-
-# 0.6.7 (May 14, 2021)
-
-### Added
-
- udp: make `UdpFramed` take `Borrow<UdpSocket>` ([#3451])
- compat: implement `AsRawFd`/`AsRawHandle` for `Compat<T>` ([#3765])
-
-[#3451]: https://github.com/tokio-rs/tokio/pull/3451
-[#3765]: https://github.com/tokio-rs/tokio/pull/3765
-
-# 0.6.6 (April 12, 2021)
-
-### Added
-
- util: makes `Framed` and `FramedStream` resumable after eof ([#3272])
- util: add `PollSemaphore::{add_permits, available_permits}` ([#3683])
-
-### Fixed
-
- chore: avoid allocation if `PollSemaphore` is unused ([#3634])
-
-[#3272]: https://github.com/tokio-rs/tokio/pull/3272
-[#3634]: https://github.com/tokio-rs/tokio/pull/3634
-[#3683]: https://github.com/tokio-rs/tokio/pull/3683
-
-# 0.6.5 (March 20, 2021)
-
-### Fixed
-
- util: annotate time module as requiring `time` feature ([#3606])
-
-[#3606]: https://github.com/tokio-rs/tokio/pull/3606
-
-# 0.6.4 (March 9, 2021)
-
-### Added
-
- codec: `AnyDelimiter` codec ([#3406])
- sync: add pollable `mpsc::Sender` ([#3490])
-
-### Fixed
-
- codec: `LinesCodec` should only return `MaxLineLengthExceeded` once per line ([#3556])
- sync: fuse PollSemaphore ([#3578])
-
-[#3406]: https://github.com/tokio-rs/tokio/pull/3406
-[#3490]: https://github.com/tokio-rs/tokio/pull/3490
-[#3556]: https://github.com/tokio-rs/tokio/pull/3556
-[#3578]: https://github.com/tokio-rs/tokio/pull/3578
-
-# 0.6.3 (January 31, 2021)
-
-### Added
-
- sync: add `ReusableBoxFuture` utility ([#3464])
-
-### Changed
-
- sync: use `ReusableBoxFuture` for `PollSemaphore` ([#3463])
- deps: remove `async-stream` dependency ([#3463])
- deps: remove `tokio-stream` dependency ([#3487])
-
-# 0.6.2 (January 21, 2021)
-
-### Added
-
- sync: add pollable `Semaphore` ([#3444])
-
-### Fixed
-
- time: fix panics on updating `DelayQueue` entries ([#3270])
-
-# 0.6.1 (January 12, 2021)
-
-### Added
-
- codec: `get_ref()`, `get_mut()`, `get_pin_mut()` and `into_inner()` for
-  `Framed`, `FramedRead`, `FramedWrite` and `StreamReader` ([#3364]).
- codec: `write_buffer()` and `write_buffer_mut()` for `Framed` and
-  `FramedWrite` ([#3387]).
-
-# 0.6.0 (December 23, 2020)
-
-### Changed
- depend on `tokio` 1.0.
-
-### Added
- rt: add constructors to `TokioContext` (#3221).
-
 # 0.5.1 (December 3, 2020)

 ### Added
@@ -221,13 +45,6 @@

 - Initial release

-[#3487]: https://github.com/tokio-rs/tokio/pull/3487
-[#3464]: https://github.com/tokio-rs/tokio/pull/3464
-[#3463]: https://github.com/tokio-rs/tokio/pull/3463
-[#3444]: https://github.com/tokio-rs/tokio/pull/3444
-[#3387]: https://github.com/tokio-rs/tokio/pull/3387
-[#3364]: https://github.com/tokio-rs/tokio/pull/3364
-[#3270]: https://github.com/tokio-rs/tokio/pull/3270
 [#2326]: https://github.com/tokio-rs/tokio/pull/2326
 [#2215]: https://github.com/tokio-rs/tokio/pull/2215
 [#2198]: https://github.com/tokio-rs/tokio/pull/2198
@@ -2,15 +2,18 @@
 name = "tokio-util"
 # When releasing to crates.io:
 # - Remove path dependencies
+# - Update html_root_url.
+# - Update doc url
+#   - Cargo.toml
 # - Update CHANGELOG.md.
-# - Create "tokio-util-0.7.x" git tag.
-version = "0.7.1"
+# - Create "v0.2.x" git tag.
+version = "0.5.1"
 edition = "2018"
-rust-version = "1.49"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
+documentation = "https://docs.rs/tokio-util/0.5.1/tokio_util"
 description = """
 Additional utilities for working with Tokio.
 """
@@ -21,36 +24,30 @@ categories = ["asynchronous"]
 default = []

 # Shorthand for enabling everything
-full = ["codec", "compat", "io-util", "time", "net", "rt"]
+full = ["codec", "compat", "io", "time", "net", "rt"]

 net = ["tokio/net"]
 compat = ["futures-io",]
-codec = ["tracing"]
+codec = ["tokio/stream"]
 time = ["tokio/time","slab"]
 io = []
-io-util = ["io", "tokio/rt", "tokio/io-util"]
-rt = ["tokio/rt", "tokio/sync", "futures-util"]
-
-__docs_rs = ["futures-util"]
+rt = ["tokio/rt"]

 [dependencies]
-tokio = { version = "1.6.0", path = "../tokio", features = ["sync"] }
+tokio = { version = "0.3.4", path = "../tokio" }

-bytes = "1.0.0"
+bytes = "0.6.0"
 futures-core = "0.3.0"
 futures-sink = "0.3.0"
 futures-io = { version = "0.3.0", optional = true }
-futures-util = { version = "0.3.0", optional = true }
+log = "0.4"
 pin-project-lite = "0.2.0"
-slab = { version = "0.4.4", optional = true } # Backs `DelayQueue`
-tracing = { version = "0.1.25", default-features = false, features = ["std"], optional = true }
+slab = { version = "0.4.1", optional = true } # Backs `DelayQueue`

 [dev-dependencies]
-tokio = { version = "1.0.0", path = "../tokio", features = ["full"] }
-tokio-test = { version = "0.4.0", path = "../tokio-test" }
-tokio-stream = { version = "0.1", path = "../tokio-stream" }
+tokio = { version = "0.3.0", path = "../tokio", features = ["full"] }
+tokio-test = { version = "0.3.0", path = "../tokio-test" }

-async-stream = "0.3.0"
 futures = "0.3.0"
 futures-test = "0.3.5"

@@ -1,4 +1,4 @@
-Copyright (c) 2022 Tokio Contributors
+Copyright (c) 2019 Tokio Contributors

 Permission is hereby granted, free of charge, to any
 person obtaining a copy of this software and associated
@@ -1,6 +1,6 @@
 # tokio-util

-Utilities for working with Tokio.
+Utilities for encoding and decoding frames.

 ## License

@@ -38,18 +38,6 @@ macro_rules! cfg_io {
    }
 }

-cfg_io! {
-    macro_rules! cfg_io_util {
-        ($($item:item)*) => {
-            $(
-                #[cfg(feature = "io-util")]
-                #[cfg_attr(docsrs, doc(cfg(feature = "io-util")))]
-                $item
-            )*
-        }
-    }
-}
-
 macro_rules! cfg_rt {
    ($($item:item)*) => {
        $(
@@ -59,13 +47,3 @@ macro_rules! cfg_rt {
        )*
    }
 }
-
-macro_rules! cfg_time {
-    ($($item:item)*) => {
-        $(
-            #[cfg(feature = "time")]
-            #[cfg_attr(docsrs, doc(cfg(feature = "time")))]
-            $item
-        )*
-    }
-}
@@ -1,263 +0,0 @@
-use crate::codec::decoder::Decoder;
-use crate::codec::encoder::Encoder;
-
-use bytes::{Buf, BufMut, Bytes, BytesMut};
-use std::{cmp, fmt, io, str, usize};
-
-const DEFAULT_SEEK_DELIMITERS: &[u8] = b",;\n\r";
-const DEFAULT_SEQUENCE_WRITER: &[u8] = b",";
-/// A simple [`Decoder`] and [`Encoder`] implementation that splits up data into chunks based on any character in the given delimiter string.
-///
-/// [`Decoder`]: crate::codec::Decoder
-/// [`Encoder`]: crate::codec::Encoder
-///
-/// # Example
-/// Decode string of bytes containing various different delimiters.
-///
-/// [`BytesMut`]: bytes::BytesMut
-/// [`Error`]: std::io::Error
-///
-/// ```
-/// use tokio_util::codec::{AnyDelimiterCodec, Decoder};
-/// use bytes::{BufMut, BytesMut};
-///
-/// #
-/// # #[tokio::main(flavor = "current_thread")]
-/// # async fn main() -> Result<(), std::io::Error> {
-/// let mut codec = AnyDelimiterCodec::new(b",;\r\n".to_vec(),b";".to_vec());
-/// let buf = &mut BytesMut::new();
-/// buf.reserve(200);
-/// buf.put_slice(b"chunk 1,chunk 2;chunk 3\n\r");
-/// assert_eq!("chunk 1", codec.decode(buf).unwrap().unwrap());
-/// assert_eq!("chunk 2", codec.decode(buf).unwrap().unwrap());
-/// assert_eq!("chunk 3", codec.decode(buf).unwrap().unwrap());
-/// assert_eq!("", codec.decode(buf).unwrap().unwrap());
-/// assert_eq!(None, codec.decode(buf).unwrap());
-/// # Ok(())
-/// # }
-/// ```
-///
-#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
-pub struct AnyDelimiterCodec {
-    // Stored index of the next index to examine for the delimiter character.
-    // This is used to optimize searching.
-    // For example, if `decode` was called with `abc` and the delimiter is '{}', it would hold `3`,
-    // because that is the next index to examine.
-    // The next time `decode` is called with `abcde}`, the method will
-    // only look at `de}` before returning.
-    next_index: usize,
-
-    /// The maximum length for a given chunk. If `usize::MAX`, chunks will be
-    /// read until a delimiter character is reached.
-    max_length: usize,
-
-    /// Are we currently discarding the remainder of a chunk which was over
-    /// the length limit?
-    is_discarding: bool,
-
-    /// The bytes that are using for search during decode
-    seek_delimiters: Vec<u8>,
-
-    /// The bytes that are using for encoding
-    sequence_writer: Vec<u8>,
-}
-
-impl AnyDelimiterCodec {
-    /// Returns a `AnyDelimiterCodec` for splitting up data into chunks.
-    ///
-    /// # Note
-    ///
-    /// The returned `AnyDelimiterCodec` will not have an upper bound on the length
-    /// of a buffered chunk. See the documentation for [`new_with_max_length`]
-    /// for information on why this could be a potential security risk.
-    ///
-    /// [`new_with_max_length`]: crate::codec::AnyDelimiterCodec::new_with_max_length()
-    pub fn new(seek_delimiters: Vec<u8>, sequence_writer: Vec<u8>) -> AnyDelimiterCodec {
-        AnyDelimiterCodec {
-            next_index: 0,
-            max_length: usize::MAX,
-            is_discarding: false,
-            seek_delimiters,
-            sequence_writer,
-        }
-    }
-
-    /// Returns a `AnyDelimiterCodec` with a maximum chunk length limit.
-    ///
-    /// If this is set, calls to `AnyDelimiterCodec::decode` will return a
-    /// [`AnyDelimiterCodecError`] when a chunk exceeds the length limit. Subsequent calls
-    /// will discard up to `limit` bytes from that chunk until a delimiter
-    /// character is reached, returning `None` until the delimiter over the limit
-    /// has been fully discarded. After that point, calls to `decode` will
-    /// function as normal.
-    ///
-    /// # Note
-    ///
-    /// Setting a length limit is highly recommended for any `AnyDelimiterCodec` which
-    /// will be exposed to untrusted input. Otherwise, the size of the buffer
-    /// that holds the chunk currently being read is unbounded. An attacker could
-    /// exploit this unbounded buffer by sending an unbounded amount of input
-    /// without any delimiter characters, causing unbounded memory consumption.
-    ///
-    /// [`AnyDelimiterCodecError`]: crate::codec::AnyDelimiterCodecError
-    pub fn new_with_max_length(
-        seek_delimiters: Vec<u8>,
-        sequence_writer: Vec<u8>,
-        max_length: usize,
-    ) -> Self {
-        AnyDelimiterCodec {
-            max_length,
-            ..AnyDelimiterCodec::new(seek_delimiters, sequence_writer)
-        }
-    }
-
-    /// Returns the maximum chunk length when decoding.
-    ///
-    /// ```
-    /// use std::usize;
-    /// use tokio_util::codec::AnyDelimiterCodec;
-    ///
-    /// let codec = AnyDelimiterCodec::new(b",;\n".to_vec(), b";".to_vec());
-    /// assert_eq!(codec.max_length(), usize::MAX);
-    /// ```
-    /// ```
-    /// use tokio_util::codec::AnyDelimiterCodec;
-    ///
-    /// let codec = AnyDelimiterCodec::new_with_max_length(b",;\n".to_vec(), b";".to_vec(), 256);
-    /// assert_eq!(codec.max_length(), 256);
-    /// ```
-    pub fn max_length(&self) -> usize {
-        self.max_length
-    }
-}
-
-impl Decoder for AnyDelimiterCodec {
-    type Item = Bytes;
-    type Error = AnyDelimiterCodecError;
-
-    fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Bytes>, AnyDelimiterCodecError> {
-        loop {
-            // Determine how far into the buffer we'll search for a delimiter. If
-            // there's no max_length set, we'll read to the end of the buffer.
-            let read_to = cmp::min(self.max_length.saturating_add(1), buf.len());
-
-            let new_chunk_offset = buf[self.next_index..read_to].iter().position(|b| {
-                self.seek_delimiters
-                    .iter()
-                    .any(|delimiter| *b == *delimiter)
-            });
-
-            match (self.is_discarding, new_chunk_offset) {
-                (true, Some(offset)) => {
-                    // If we found a new chunk, discard up to that offset and
-                    // then stop discarding. On the next iteration, we'll try
-                    // to read a chunk normally.
-                    buf.advance(offset + self.next_index + 1);
-                    self.is_discarding = false;
-                    self.next_index = 0;
-                }
-                (true, None) => {
-                    // Otherwise, we didn't find a new chunk, so we'll discard
-                    // everything we read. On the next iteration, we'll continue
-                    // discarding up to max_len bytes unless we find a new chunk.
-                    buf.advance(read_to);
-                    self.next_index = 0;
-                    if buf.is_empty() {
-                        return Ok(None);
-                    }
-                }
-                (false, Some(offset)) => {
-                    // Found a chunk!
-                    let new_chunk_index = offset + self.next_index;
-                    self.next_index = 0;
-                    let mut chunk = buf.split_to(new_chunk_index + 1);
-                    chunk.truncate(chunk.len() - 1);
-                    let chunk = chunk.freeze();
-                    return Ok(Some(chunk));
-                }
-                (false, None) if buf.len() > self.max_length => {
-                    // Reached the maximum length without finding a
-                    // new chunk, return an error and start discarding on the
-                    // next call.
-                    self.is_discarding = true;
-                    return Err(AnyDelimiterCodecError::MaxChunkLengthExceeded);
-                }
-                (false, None) => {
-                    // We didn't find a chunk or reach the length limit, so the next
-                    // call will resume searching at the current offset.
-                    self.next_index = read_to;
-                    return Ok(None);
-                }
-            }
-        }
-    }
-
-    fn decode_eof(&mut self, buf: &mut BytesMut) -> Result<Option<Bytes>, AnyDelimiterCodecError> {
-        Ok(match self.decode(buf)? {
-            Some(frame) => Some(frame),
-            None => {
-                // return remaining data, if any
-                if buf.is_empty() {
-                    None
-                } else {
-                    let chunk = buf.split_to(buf.len());
-                    self.next_index = 0;
-                    Some(chunk.freeze())
-                }
-            }
-        })
-    }
-}
-
-impl<T> Encoder<T> for AnyDelimiterCodec
-where
-    T: AsRef<str>,
-{
-    type Error = AnyDelimiterCodecError;
-
-    fn encode(&mut self, chunk: T, buf: &mut BytesMut) -> Result<(), AnyDelimiterCodecError> {
-        let chunk = chunk.as_ref();
-        buf.reserve(chunk.len() + 1);
-        buf.put(chunk.as_bytes());
-        buf.put(self.sequence_writer.as_ref());
-
-        Ok(())
-    }
-}
-
-impl Default for AnyDelimiterCodec {
-    fn default() -> Self {
-        Self::new(
-            DEFAULT_SEEK_DELIMITERS.to_vec(),
-            DEFAULT_SEQUENCE_WRITER.to_vec(),
-        )
-    }
-}
-
-/// An error occurred while encoding or decoding a chunk.
-#[derive(Debug)]
-pub enum AnyDelimiterCodecError {
-    /// The maximum chunk length was exceeded.
-    MaxChunkLengthExceeded,
-    /// An IO error occurred.
-    Io(io::Error),
-}
-
-impl fmt::Display for AnyDelimiterCodecError {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            AnyDelimiterCodecError::MaxChunkLengthExceeded => {
-                write!(f, "max chunk length exceeded")
-            }
-            AnyDelimiterCodecError::Io(e) => write!(f, "{}", e),
-        }
-    }
-}
-
-impl From<io::Error> for AnyDelimiterCodecError {
-    fn from(e: io::Error) -> AnyDelimiterCodecError {
-        AnyDelimiterCodecError::Io(e)
-    }
-}
-
-impl std::error::Error for AnyDelimiterCodecError {}
@@ -74,13 +74,3 @@ impl Encoder<Bytes> for BytesCodec {
        Ok(())
    }
 }
-
-impl Encoder<BytesMut> for BytesCodec {
-    type Error = io::Error;
-
-    fn encode(&mut self, data: BytesMut, buf: &mut BytesMut) -> Result<(), io::Error> {
-        buf.reserve(data.len());
-        buf.put(data);
-        Ok(())
-    }
-}
@@ -16,20 +16,6 @@ use std::io;
 /// implementing stateful streaming parsers. In many cases, though, this type
 /// will simply be a unit struct (e.g. `struct HttpDecoder`).
 ///
-/// For some underlying data-sources, namely files and FIFOs,
-/// it's possible to temporarily read 0 bytes by reaching EOF.
-///
-/// In these cases `decode_eof` will be called until it signals
-/// fullfillment of all closing frames by returning `Ok(None)`.
-/// After that, repeated attempts to read from the [`Framed`] or [`FramedRead`]
-/// will not invoke `decode` or `decode_eof` again, until data can be read
-/// during a retry.
-///
-/// It is up to the Decoder to keep track of a restart after an EOF,
-/// and to decide how to handle such an event by, for example,
-/// allowing frames to cross EOF boundaries, re-emitting opening frames, or
-/// resetting the entire internal state.
-///
 /// [`Framed`]: crate::codec::Framed
 /// [`FramedRead`]: crate::codec::FramedRead
 pub trait Decoder {
@@ -129,18 +115,13 @@ pub trait Decoder {
    /// This method defaults to calling `decode` and returns an error if
    /// `Ok(None)` is returned while there is unconsumed data in `buf`.
    /// Typically this doesn't need to be implemented unless the framing
-    /// protocol differs near the end of the stream, or if you need to construct
-    /// frames _across_ eof boundaries on sources that can be resumed.
+    /// protocol differs near the end of the stream.
    ///
    /// Note that the `buf` argument may be empty. If a previous call to
    /// `decode_eof` consumed all the bytes in the buffer, `decode_eof` will be
    /// called again until it returns `None`, indicating that there are no more
    /// frames to yield. This behavior enables returning finalization frames
    /// that may not be based on inbound data.
-    ///
-    /// Once `None` has been returned, `decode_eof` won't be called again until
-    /// an attempt to resume the stream has been made, where the underlying stream
-    /// actually returned more data.
    fn decode_eof(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        match self.decode(buf)? {
            Some(frame) => Ok(Some(frame)),
@@ -172,7 +153,7 @@ pub trait Decoder {
    /// calling `split` on the [`Framed`] returned by this method, which will
    /// break them into separate objects, allowing them to interact more easily.
    ///
-    /// [`Stream`]: futures_core::Stream
+    /// [`Stream`]: tokio::stream::Stream
    /// [`Sink`]: futures_sink::Sink
    /// [`Framed`]: crate::codec::Framed
    fn framed<T: AsyncRead + AsyncWrite + Sized>(self, io: T) -> Framed<T, Self>
@@ -2,8 +2,10 @@ use crate::codec::decoder::Decoder;
 use crate::codec::encoder::Encoder;
 use crate::codec::framed_impl::{FramedImpl, RWFrames, ReadFrame, WriteFrame};

-use futures_core::Stream;
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::{
+    io::{AsyncRead, AsyncWrite},
+    stream::Stream,
+};

 use bytes::BytesMut;
 use futures_sink::Sink;
@@ -20,7 +22,7 @@ pin_project! {
    /// You can create a `Framed` instance by using the [`Decoder::framed`] adapter, or
    /// by using the `new` function seen below.
    ///
-    /// [`Stream`]: futures_core::Stream
+    /// [`Stream`]: tokio::stream::Stream
    /// [`Sink`]: futures_sink::Sink
    /// [`AsyncRead`]: tokio::io::AsyncRead
    /// [`Decoder::framed`]: crate::codec::Decoder::framed()
@@ -52,12 +54,7 @@ where
    /// calling [`split`] on the `Framed` returned by this method, which will
    /// break them into separate objects, allowing them to interact more easily.
    ///
-    /// Note that, for some byte sources, the stream can be resumed after an EOF
-    /// by reading from it, even after it has returned `None`. Repeated attempts
-    /// to do so, without new data available, continue to return `None` without
-    /// creating more (closing) frames.
-    ///
-    /// [`Stream`]: futures_core::Stream
+    /// [`Stream`]: tokio::stream::Stream
    /// [`Sink`]: futures_sink::Sink
    /// [`Decode`]: crate::codec::Decoder
    /// [`Encoder`]: crate::codec::Encoder
@@ -91,7 +88,7 @@ where
    /// calling [`split`] on the `Framed` returned by this method, which will
    /// break them into separate objects, allowing them to interact more easily.
    ///
-    /// [`Stream`]: futures_core::Stream
+    /// [`Stream`]: tokio::stream::Stream
    /// [`Sink`]: futures_sink::Sink
    /// [`Decode`]: crate::codec::Decoder
    /// [`Encoder`]: crate::codec::Encoder
@@ -106,7 +103,6 @@ where
                        eof: false,
                        is_readable: false,
                        buffer: BytesMut::with_capacity(capacity),
-                        has_errored: false,
                    },
                    write: WriteFrame::default(),
                },
@@ -137,7 +133,7 @@ impl<T, U> Framed<T, U> {
    /// calling [`split`] on the `Framed` returned by this method, which will
    /// break them into separate objects, allowing them to interact more easily.
    ///
-    /// [`Stream`]: futures_core::Stream
+    /// [`Stream`]: tokio::stream::Stream
    /// [`Sink`]: futures_sink::Sink
    /// [`Decoder`]: crate::codec::Decoder
    /// [`Encoder`]: crate::codec::Encoder
@@ -176,16 +172,6 @@ impl<T, U> Framed<T, U> {
        &mut self.inner.inner
    }

-    /// Returns a pinned mutable reference to the underlying I/O stream wrapped by
-    /// `Framed`.
-    ///
-    /// Note that care should be taken to not tamper with the underlying stream
-    /// of data coming in as it may corrupt the stream of frames otherwise
-    /// being worked with.
-    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut T> {
-        self.project().inner.project().inner
-    }
-
    /// Returns a reference to the underlying codec wrapped by
    /// `Framed`.
    ///
@@ -204,35 +190,6 @@ impl<T, U> Framed<T, U> {
        &mut self.inner.codec
    }

-    /// Maps the codec `U` to `C`, preserving the read and write buffers
-    /// wrapped by `Framed`.
-    ///
-    /// Note that care should be taken to not tamper with the underlying codec
-    /// as it may corrupt the stream of frames otherwise being worked with.
-    pub fn map_codec<C, F>(self, map: F) -> Framed<T, C>
-    where
-        F: FnOnce(U) -> C,
-    {
-        // This could be potentially simplified once rust-lang/rust#86555 hits stable
-        let parts = self.into_parts();
-        Framed::from_parts(FramedParts {
-            io: parts.io,
-            codec: map(parts.codec),
-            read_buf: parts.read_buf,
-            write_buf: parts.write_buf,
-            _priv: (),
-        })
-    }
-
-    /// Returns a mutable reference to the underlying codec wrapped by
-    /// `Framed`.
-    ///
-    /// Note that care should be taken to not tamper with the underlying codec
-    /// as it may corrupt the stream of frames otherwise being worked with.
-    pub fn codec_pin_mut(self: Pin<&mut Self>) -> &mut U {
-        self.project().inner.project().codec
-    }
-
    /// Returns a reference to the read buffer.
    pub fn read_buffer(&self) -> &BytesMut {
        &self.inner.state.read.buffer
@@ -243,16 +200,6 @@ impl<T, U> Framed<T, U> {
        &mut self.inner.state.read.buffer
    }

-    /// Returns a reference to the write buffer.
-    pub fn write_buffer(&self) -> &BytesMut {
-        &self.inner.state.write.buffer
-    }
-
-    /// Returns a mutable reference to the write buffer.
-    pub fn write_buffer_mut(&mut self) -> &mut BytesMut {
-        &mut self.inner.state.write.buffer
-    }
-
    /// Consumes the `Framed`, returning its underlying I/O stream.
    ///
    /// Note that care should be taken to not tamper with the underlying stream
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
Alice Ryhl	171724863d	chore: prepare tokio-macros v0.3.2 (#3295 )	2020-12-19 17:03:42 +01:00
Alice Ryhl	1e880645a0	chore: fix stress test (#3296 )	2020-12-19 12:11:22 +01:00
bdonlan	176c809e20	chore: prepare v0.3.6 release (#3276 )	2020-12-14 13:51:38 -08:00
Alice Ryhland@tijsvd	d2676db20c	watch: fix spurious wakeup (#3244 ) Co-authored-by: @tijsvd	2020-12-10 12:00:50 +01:00
Alice Ryhl	60f16313be	ci: enable ci for v0.3.x (#3245 )	2020-12-10 11:30:09 +01:00