From 8efa62013b551d5130791c3a79ce8ab5cb0b5abf Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Tue, 15 Dec 2020 23:24:38 -0500 Subject: [PATCH] Move stream items into `tokio-stream` (#3277) This change removes all references to `Stream` from within the `tokio` crate and moves them into a new `tokio-stream` crate. Most types have had their `impl Stream` removed as well in-favor of their inherent methods. Closes #2870 --- CONTRIBUTING.md | 1 - Cargo.toml | 1 + benches/sync_mpsc.rs | 12 +- examples/Cargo.toml | 5 +- examples/chat.rs | 15 +- examples/print_each_packet.rs | 2 +- examples/tinydb.rs | 2 +- examples/tinyhttp.rs | 2 +- examples/udp-codec.rs | 2 +- tokio-stream/Cargo.toml | 38 +++++ {tokio/src/stream => tokio-stream/src}/all.rs | 2 +- {tokio/src/stream => tokio-stream/src}/any.rs | 2 +- .../src/stream => tokio-stream/src}/chain.rs | 2 +- .../stream => tokio-stream/src}/collect.rs | 4 +- .../src/stream => tokio-stream/src}/empty.rs | 4 +- .../src/stream => tokio-stream/src}/filter.rs | 2 +- .../stream => tokio-stream/src}/filter_map.rs | 2 +- .../src/stream => tokio-stream/src}/fold.rs | 2 +- .../src/stream => tokio-stream/src}/fuse.rs | 2 +- .../src/stream => tokio-stream/src}/iter.rs | 21 ++- .../stream/mod.rs => tokio-stream/src/lib.rs | 116 +++++++++---- tokio-stream/src/macros.rs | 18 ++ {tokio/src/stream => tokio-stream/src}/map.rs | 2 +- .../src/stream => tokio-stream/src}/merge.rs | 2 +- .../src/stream => tokio-stream/src}/next.rs | 2 +- .../src/stream => tokio-stream/src}/once.rs | 6 +- .../stream => tokio-stream/src}/pending.rs | 6 +- .../src/stream => tokio-stream/src}/skip.rs | 2 +- .../stream => tokio-stream/src}/skip_while.rs | 2 +- .../stream => tokio-stream/src}/stream_map.rs | 157 +++++++++++++++--- .../src/stream => tokio-stream/src}/take.rs | 2 +- .../stream => tokio-stream/src}/take_while.rs | 2 +- .../stream => tokio-stream/src}/throttle.rs | 6 +- .../stream => tokio-stream/src}/timeout.rs | 33 +++- .../stream => tokio-stream/src}/try_next.rs | 2 +- tokio-stream/tests/async_send_sync.rs | 105 ++++++++++++ {tokio => tokio-stream}/tests/stream_chain.rs | 15 +- .../tests/stream_collect.rs | 23 ++- {tokio => tokio-stream}/tests/stream_empty.rs | 2 +- {tokio => tokio-stream}/tests/stream_fuse.rs | 2 +- {tokio => tokio-stream}/tests/stream_iter.rs | 2 +- {tokio => tokio-stream}/tests/stream_merge.rs | 15 +- {tokio => tokio-stream}/tests/stream_once.rs | 2 +- .../tests/stream_pending.rs | 2 +- .../tests/stream_stream_map.rs | 30 +++- .../tests/stream_timeout.rs | 2 +- tokio-stream/tests/support/mpsc.rs | 15 ++ .../tests/time_throttle.rs | 2 +- tokio-test/Cargo.toml | 2 + tokio-test/src/io.rs | 23 ++- tokio-test/src/task.rs | 2 +- tokio-util/Cargo.toml | 1 + tokio-util/src/codec/decoder.rs | 2 +- tokio-util/src/codec/framed.rs | 14 +- tokio-util/src/codec/framed_impl.rs | 6 +- tokio-util/src/codec/framed_read.rs | 5 +- tokio-util/src/codec/framed_write.rs | 3 +- tokio-util/src/codec/mod.rs | 2 +- tokio-util/src/either.rs | 6 +- tokio-util/src/io/read_buf.rs | 2 +- tokio-util/src/io/reader_stream.rs | 6 +- tokio-util/src/io/stream_reader.rs | 4 +- tokio-util/src/lib.rs | 2 +- tokio-util/src/udp/frame.rs | 5 +- tokio-util/tests/framed.rs | 3 +- tokio-util/tests/io_reader_stream.rs | 2 +- tokio-util/tests/io_stream_reader.rs | 2 +- tokio-util/tests/udp.rs | 3 +- tokio/Cargo.toml | 2 + tokio/src/coop.rs | 4 +- tokio/src/fs/read_dir.rs | 16 +- tokio/src/io/mod.rs | 2 +- tokio/src/io/util/async_buf_read_ext.rs | 11 +- tokio/src/io/util/lines.rs | 13 -- tokio/src/io/util/split.rs | 13 -- tokio/src/lib.rs | 11 +- tokio/src/macros/cfg.rs | 11 -- tokio/src/macros/pin.rs | 2 +- tokio/src/macros/select.rs | 6 +- tokio/src/net/tcp/listener.rs | 34 +--- tokio/src/net/udp.rs | 3 +- tokio/src/net/unix/listener.rs | 24 +-- tokio/src/signal/unix.rs | 10 -- tokio/src/signal/windows.rs | 23 +-- tokio/src/sync/broadcast.rs | 67 -------- tokio/src/sync/mpsc/bounded.rs | 12 +- tokio/src/sync/mpsc/mod.rs | 7 +- tokio/src/sync/mpsc/unbounded.rs | 9 - tokio/src/time/interval.rs | 14 -- tokio/src/util/mod.rs | 4 +- tokio/src/util/rand.rs | 2 +- tokio/tests/async_send_sync.rs | 14 +- tokio/tests/fs_dir.rs | 32 ---- tokio/tests/io_lines.rs | 16 -- tokio/tests/rt_basic.rs | 15 +- tokio/tests/support/mpsc_stream.rs | 29 ++++ tokio/tests/sync_broadcast.rs | 40 ----- tokio/tests/sync_mpsc.rs | 15 +- tokio/tests/task_blocking.rs | 14 +- tokio/tests/tcp_accept.rs | 4 +- tokio/tests/time_interval.rs | 15 -- 101 files changed, 710 insertions(+), 577 deletions(-) create mode 100644 tokio-stream/Cargo.toml rename {tokio/src/stream => tokio-stream/src}/all.rs (98%) rename {tokio/src/stream => tokio-stream/src}/any.rs (98%) rename {tokio/src/stream => tokio-stream/src}/chain.rs (96%) rename {tokio/src/stream => tokio-stream/src}/collect.rs (98%) rename {tokio/src/stream => tokio-stream/src}/empty.rs (93%) rename {tokio/src/stream => tokio-stream/src}/filter.rs (98%) rename {tokio/src/stream => tokio-stream/src}/filter_map.rs (98%) rename {tokio/src/stream => tokio-stream/src}/fold.rs (98%) rename {tokio/src/stream => tokio-stream/src}/fuse.rs (97%) rename {tokio/src/stream => tokio-stream/src}/iter.rs (74%) rename tokio/src/stream/mod.rs => tokio-stream/src/lib.rs (89%) create mode 100644 tokio-stream/src/macros.rs rename {tokio/src/stream => tokio-stream/src}/map.rs (97%) rename {tokio/src/stream => tokio-stream/src}/merge.rs (98%) rename {tokio/src/stream => tokio-stream/src}/next.rs (97%) rename {tokio/src/stream => tokio-stream/src}/once.rs (88%) rename {tokio/src/stream => tokio-stream/src}/pending.rs (89%) rename {tokio/src/stream => tokio-stream/src}/skip.rs (98%) rename {tokio/src/stream => tokio-stream/src}/skip_while.rs (98%) rename {tokio/src/stream => tokio-stream/src}/stream_map.rs (77%) rename {tokio/src/stream => tokio-stream/src}/take.rs (98%) rename {tokio/src/stream => tokio-stream/src}/take_while.rs (98%) rename {tokio/src/stream => tokio-stream/src}/throttle.rs (95%) rename {tokio/src/stream => tokio-stream/src}/timeout.rs (72%) rename {tokio/src/stream => tokio-stream/src}/try_next.rs (96%) create mode 100644 tokio-stream/tests/async_send_sync.rs rename {tokio => tokio-stream}/tests/stream_chain.rs (89%) rename {tokio => tokio-stream}/tests/stream_collect.rs (89%) rename {tokio => tokio-stream}/tests/stream_empty.rs (79%) rename {tokio => tokio-stream}/tests/stream_fuse.rs (96%) rename {tokio => tokio-stream}/tests/stream_iter.rs (91%) rename {tokio => tokio-stream}/tests/stream_merge.rs (86%) rename {tokio => tokio-stream}/tests/stream_once.rs (82%) rename {tokio => tokio-stream}/tests/stream_pending.rs (85%) rename {tokio => tokio-stream}/tests/stream_stream_map.rs (93%) rename {tokio => tokio-stream}/tests/stream_timeout.rs (98%) create mode 100644 tokio-stream/tests/support/mpsc.rs rename {tokio => tokio-stream}/tests/time_throttle.rs (95%) create mode 100644 tokio/tests/support/mpsc_stream.rs diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 18e38e808..7de50fc6a 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -486,7 +486,6 @@ missing a difficulty rating, and you should feel free to add one. - **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. diff --git a/Cargo.toml b/Cargo.toml index 1867acb72..bc01f1862 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ members = [ "tokio", "tokio-macros", "tokio-test", + "tokio-stream", "tokio-util", # Internal diff --git a/benches/sync_mpsc.rs b/benches/sync_mpsc.rs index 3f7e3fcaa..361493b12 100644 --- a/benches/sync_mpsc.rs +++ b/benches/sync_mpsc.rs @@ -30,22 +30,26 @@ 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::(1000); - let _ = tx.try_send([0; 64]); + let _ = rt.block_on(tx.send([0; 64])); - rx.try_recv().unwrap(); + rt.block_on(rx.recv()).unwrap(); }); } fn send_large(b: &mut Bencher) { + let rt = rt(); + b.iter(|| { let (tx, mut rx) = mpsc::channel::(1000); - let _ = tx.try_send([[0; 64]; 64]); + let _ = rt.block_on(tx.send([[0; 64]; 64])); - rx.try_recv().unwrap(); + rt.block_on(rx.recv()).unwrap(); }); } diff --git a/examples/Cargo.toml b/examples/Cargo.toml index 4b7362482..fc35e60f3 100644 --- a/examples/Cargo.toml +++ b/examples/Cargo.toml @@ -8,9 +8,12 @@ edition = "2018" # [dependencies] instead. [dev-dependencies] tokio = { version = "1.0.0", path = "../tokio", features = ["full", "tracing"] } +tokio-util = { version = "0.6.0", path = "../tokio-util", features = ["full"] } +tokio-stream = { version = "0.1", path = "../tokio-stream" } + +async-stream = "0.3" tracing = "0.1" tracing-subscriber = { version = "0.2.7", default-features = false, features = ["fmt", "ansi", "env-filter", "chrono", "tracing-log"] } -tokio-util = { version = "0.6.0", path = "../tokio-util", features = ["full"] } bytes = "0.6" futures = "0.3.0" http = "0.2" diff --git a/examples/chat.rs b/examples/chat.rs index 3f9450392..821fd964b 100644 --- a/examples/chat.rs +++ b/examples/chat.rs @@ -27,8 +27,8 @@ #![warn(rust_2018_idioms)] use tokio::net::{TcpListener, TcpStream}; -use tokio::stream::{Stream, StreamExt}; use tokio::sync::{mpsc, Mutex}; +use tokio_stream::{Stream, StreamExt}; use tokio_util::codec::{Framed, LinesCodec, LinesCodecError}; use futures::SinkExt; @@ -101,9 +101,6 @@ async fn main() -> Result<(), Box> { /// Shorthand for the transmit half of the message channel. type Tx = mpsc::UnboundedSender; -/// Shorthand for the receive half of the message channel. -type Rx = mpsc::UnboundedReceiver; - /// Data that is shared between all peers in the chat server. /// /// This is the set of `Tx` handles for all connected clients. Whenever a @@ -127,7 +124,7 @@ struct Peer { /// /// This is used to receive messages from peers. When a message is received /// off of this `Rx`, it will be written to the socket. - rx: Rx, + rx: Pin + Send>>, } impl Shared { @@ -159,11 +156,17 @@ impl Peer { let addr = lines.get_ref().peer_addr()?; // Create a channel for this peer - let (tx, rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = mpsc::unbounded_channel(); // Add an entry for this `Peer` in the shared state map. state.lock().await.peers.insert(addr, tx); + let rx = Box::pin(async_stream::stream! { + while let Some(item) = rx.recv().await { + yield item; + } + }); + Ok(Peer { lines, rx }) } } diff --git a/examples/print_each_packet.rs b/examples/print_each_packet.rs index b3e1b17ec..087f9cf03 100644 --- a/examples/print_each_packet.rs +++ b/examples/print_each_packet.rs @@ -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; diff --git a/examples/tinydb.rs b/examples/tinydb.rs index f0db7fa86..9da429ace 100644 --- a/examples/tinydb.rs +++ b/examples/tinydb.rs @@ -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; diff --git a/examples/tinyhttp.rs b/examples/tinyhttp.rs index c561bbd31..e86305f36 100644 --- a/examples/tinyhttp.rs +++ b/examples/tinyhttp.rs @@ -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] diff --git a/examples/udp-codec.rs b/examples/udp-codec.rs index 8b64cbc3a..7c305245b 100644 --- a/examples/udp-codec.rs +++ b/examples/udp-codec.rs @@ -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; diff --git a/tokio-stream/Cargo.toml b/tokio-stream/Cargo.toml new file mode 100644 index 000000000..b93d4d568 --- /dev/null +++ b/tokio-stream/Cargo.toml @@ -0,0 +1,38 @@ +[package] +name = "tokio-stream" +# When releasing to crates.io: +# - Remove path dependencies +# - Update html_root_url. +# - Update doc url +# - Cargo.toml +# - Update CHANGELOG.md. +# - Create "tokio-stream-0.1.x" git tag. +version = "0.1.0" +edition = "2018" +authors = ["Tokio Contributors "] +license = "MIT" +repository = "https://github.com/tokio-rs/tokio" +homepage = "https://tokio.rs" +documentation = "https://docs.rs/tokio-stream/0.1.0/tokio_stream" +description = """ +Utilities to work with `Stream` and `tokio`. +""" +categories = ["asynchronous"] +publish = false + +[features] +default = ["time"] +time = ["tokio/time"] + +[dependencies] +futures-core = { version = "0.3.0" } +pin-project-lite = "0.2.0" +tokio = { version = "1.0", path = "../tokio", features = ["sync"] } +async-stream = "0.3" + +[dev-dependencies] +tokio = { version = "1.0", path = "../tokio", features = ["full"] } +tokio-test = { path = "../tokio-test" } +futures = { version = "0.3", default-features = false } + +proptest = "0.10.0" \ No newline at end of file diff --git a/tokio/src/stream/all.rs b/tokio-stream/src/all.rs similarity index 98% rename from tokio/src/stream/all.rs rename to tokio-stream/src/all.rs index 353d61a3b..11573f9b9 100644 --- a/tokio/src/stream/all.rs +++ b/tokio-stream/src/all.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::future::Future; use core::marker::PhantomPinned; diff --git a/tokio/src/stream/any.rs b/tokio-stream/src/any.rs similarity index 98% rename from tokio/src/stream/any.rs rename to tokio-stream/src/any.rs index aac0ec756..4c4c59394 100644 --- a/tokio/src/stream/any.rs +++ b/tokio-stream/src/any.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::future::Future; use core::marker::PhantomPinned; diff --git a/tokio/src/stream/chain.rs b/tokio-stream/src/chain.rs similarity index 96% rename from tokio/src/stream/chain.rs rename to tokio-stream/src/chain.rs index 6124c91e4..cfdef83d7 100644 --- a/tokio/src/stream/chain.rs +++ b/tokio-stream/src/chain.rs @@ -1,4 +1,4 @@ -use crate::stream::{Fuse, Stream}; +use crate::{Fuse, Stream}; use core::pin::Pin; use core::task::{Context, Poll}; diff --git a/tokio/src/stream/collect.rs b/tokio-stream/src/collect.rs similarity index 98% rename from tokio/src/stream/collect.rs rename to tokio-stream/src/collect.rs index 1aafc303d..23f48b049 100644 --- a/tokio/src/stream/collect.rs +++ b/tokio-stream/src/collect.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::future::Future; use core::marker::PhantomPinned; @@ -26,7 +26,7 @@ pin_project! { } } -/// Convert from a [`Stream`](crate::stream::Stream). +/// Convert from a [`Stream`](crate::Stream). /// /// This trait is not intended to be used directly. Instead, call /// [`StreamExt::collect()`](super::StreamExt::collect). diff --git a/tokio/src/stream/empty.rs b/tokio-stream/src/empty.rs similarity index 93% rename from tokio/src/stream/empty.rs rename to tokio-stream/src/empty.rs index 2f56ac6ca..965dcf5da 100644 --- a/tokio/src/stream/empty.rs +++ b/tokio-stream/src/empty.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::marker::PhantomData; use core::pin::Pin; @@ -24,7 +24,7 @@ unsafe impl Sync for Empty {} /// Basic usage: /// /// ``` -/// use tokio::stream::{self, StreamExt}; +/// use tokio_stream::{self as stream, StreamExt}; /// /// #[tokio::main] /// async fn main() { diff --git a/tokio/src/stream/filter.rs b/tokio-stream/src/filter.rs similarity index 98% rename from tokio/src/stream/filter.rs rename to tokio-stream/src/filter.rs index 799630b23..f3dd8716b 100644 --- a/tokio/src/stream/filter.rs +++ b/tokio-stream/src/filter.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::fmt; use core::pin::Pin; diff --git a/tokio/src/stream/filter_map.rs b/tokio-stream/src/filter_map.rs similarity index 98% rename from tokio/src/stream/filter_map.rs rename to tokio-stream/src/filter_map.rs index 8dc05a546..fe604a6f4 100644 --- a/tokio/src/stream/filter_map.rs +++ b/tokio-stream/src/filter_map.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::fmt; use core::pin::Pin; diff --git a/tokio/src/stream/fold.rs b/tokio-stream/src/fold.rs similarity index 98% rename from tokio/src/stream/fold.rs rename to tokio-stream/src/fold.rs index 5cf2bfafa..e2e97d8f3 100644 --- a/tokio/src/stream/fold.rs +++ b/tokio-stream/src/fold.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::future::Future; use core::marker::PhantomPinned; diff --git a/tokio/src/stream/fuse.rs b/tokio-stream/src/fuse.rs similarity index 97% rename from tokio/src/stream/fuse.rs rename to tokio-stream/src/fuse.rs index 6c9e02d66..2500641d9 100644 --- a/tokio/src/stream/fuse.rs +++ b/tokio-stream/src/fuse.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use pin_project_lite::pin_project; use std::pin::Pin; diff --git a/tokio/src/stream/iter.rs b/tokio-stream/src/iter.rs similarity index 74% rename from tokio/src/stream/iter.rs rename to tokio-stream/src/iter.rs index bc0388a14..128be616f 100644 --- a/tokio/src/stream/iter.rs +++ b/tokio-stream/src/iter.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::pin::Pin; use core::task::{Context, Poll}; @@ -8,6 +8,7 @@ use core::task::{Context, Poll}; #[must_use = "streams do nothing unless polled"] pub struct Iter { iter: I, + yield_amt: usize, } impl Unpin for Iter {} @@ -20,7 +21,7 @@ impl Unpin for Iter {} /// /// ``` /// # async fn dox() { -/// use tokio::stream::{self, StreamExt}; +/// use tokio_stream::{self as stream, StreamExt}; /// /// let mut stream = stream::iter(vec![17, 19]); /// @@ -35,6 +36,7 @@ where { Iter { iter: i.into_iter(), + yield_amt: 0, } } @@ -45,9 +47,18 @@ where type Item = I::Item; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let coop = ready!(crate::coop::poll_proceed(cx)); - coop.made_progress(); - Poll::Ready(self.iter.next()) + // TODO: add coop back + if self.yield_amt >= 32 { + self.yield_amt = 0; + + cx.waker().wake_by_ref(); + + Poll::Pending + } else { + self.yield_amt += 1; + + Poll::Ready(self.iter.next()) + } } fn size_hint(&self) -> (usize, Option) { diff --git a/tokio/src/stream/mod.rs b/tokio-stream/src/lib.rs similarity index 89% rename from tokio/src/stream/mod.rs rename to tokio-stream/src/lib.rs index 81afe7aef..ae8b2e2e4 100644 --- a/tokio/src/stream/mod.rs +++ b/tokio-stream/src/lib.rs @@ -1,3 +1,28 @@ +#![doc(html_root_url = "https://docs.rs/tokio-stream/0.1.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, deny(broken_intra_doc_links))] +#![doc(test( + no_crate_inject, + attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables)) +))] +#![cfg_attr(docsrs, feature(doc_cfg))] +#![cfg_attr(docsrs, deny(broken_intra_doc_links))] +#![doc(test( + no_crate_inject, + attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables)) +))] +#![cfg_attr(docsrs, feature(doc_cfg))] + //! Stream utilities for Tokio. //! //! A `Stream` is an asynchronous sequence of values. It can be thought of as @@ -15,7 +40,7 @@ //! `while let` loop as follows: //! //! ```rust -//! use tokio::stream::{self, StreamExt}; +//! use tokio_stream::{self as stream, StreamExt}; //! //! #[tokio::main] //! async fn main() { @@ -46,13 +71,16 @@ //! [`tokio-util`] provides the [`StreamReader`] and [`ReaderStream`] //! types when the io feature is enabled. //! -//! [tokio-util]: https://docs.rs/tokio-util/0.3/tokio_util/codec/index.html -//! [`tokio::io`]: crate::io -//! [`AsyncRead`]: crate::io::AsyncRead -//! [`AsyncWrite`]: crate::io::AsyncWrite +//! [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; + mod all; use all::AllFuture; @@ -120,9 +148,9 @@ use take_while::TakeWhile; cfg_time! { mod timeout; use timeout::Timeout; - use crate::time::Duration; + use tokio::time::Duration; mod throttle; - use crate::stream::throttle::{throttle, Throttle}; + use crate::throttle::{throttle, Throttle}; } #[doc(no_inline)] @@ -146,11 +174,11 @@ pub use futures_core::Stream; /// # #[tokio::main(flavor = "current_thread")] /// # async fn main() { /// -/// let a = tokio::stream::iter(vec![1, 3, 5]); -/// let b = tokio::stream::iter(vec![2, 4, 6]); +/// let a = tokio_stream::iter(vec![1, 3, 5]); +/// let b = tokio_stream::iter(vec![2, 4, 6]); /// /// // use the fully qualified call syntax for the other trait: -/// let merged = tokio::stream::StreamExt::merge(a, b); +/// let merged = tokio_stream::StreamExt::merge(a, b); /// /// // use normal call notation for futures::stream::StreamExt::collect /// let output: Vec<_> = merged.collect().await; @@ -158,7 +186,7 @@ pub use futures_core::Stream; /// # } /// ``` /// -/// [`Stream`]: crate::stream::Stream +/// [`Stream`]: crate::Stream /// [futures]: https://docs.rs/futures /// [futures-StreamExt]: https://docs.rs/futures/0.3/futures/stream/trait.StreamExt.html pub trait StreamExt: Stream { @@ -183,7 +211,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// let mut stream = stream::iter(1..=3); /// @@ -219,7 +247,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// let mut stream = stream::iter(vec![Ok(1), Ok(2), Err("nope")]); /// @@ -251,7 +279,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// let stream = stream::iter(1..=3); /// let mut stream = stream.map(|x| x + 3); @@ -284,16 +312,17 @@ pub trait StreamExt: Stream { /// /// For merging multiple streams, consider using [`StreamMap`] instead. /// - /// [`StreamMap`]: crate::stream::StreamMap + /// [`StreamMap`]: crate::StreamMap /// /// # Examples /// /// ``` - /// use tokio::stream::StreamExt; + /// use tokio_stream::{StreamExt, Stream}; /// use tokio::sync::mpsc; /// use tokio::time; /// /// use std::time::Duration; + /// use std::pin::Pin; /// /// # /* /// #[tokio::main] @@ -301,8 +330,21 @@ pub trait StreamExt: Stream { /// # #[tokio::main(flavor = "current_thread")] /// async fn main() { /// # time::pause(); - /// let (tx1, rx1) = mpsc::channel(10); - /// let (tx2, rx2) = mpsc::channel(10); + /// let (tx1, mut rx1) = mpsc::channel::(10); + /// let (tx2, mut rx2) = mpsc::channel::(10); + /// + /// // Convert the channels to a `Stream`. + /// let rx1 = Box::pin(async_stream::stream! { + /// while let Some(item) = rx1.recv().await { + /// yield item; + /// } + /// }) as Pin + Send>>; + /// + /// let rx2 = Box::pin(async_stream::stream! { + /// while let Some(item) = rx2.recv().await { + /// yield item; + /// } + /// }) as Pin + Send>>; /// /// let mut rx = rx1.merge(rx2); /// @@ -365,7 +407,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// let stream = stream::iter(1..=8); /// let mut evens = stream.filter(|x| x % 2 == 0); @@ -401,7 +443,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// let stream = stream::iter(1..=8); /// let mut evens = stream.filter_map(|x| { @@ -433,7 +475,7 @@ pub trait StreamExt: Stream { /// # Examples /// /// ``` - /// use tokio::stream::{Stream, StreamExt}; + /// use tokio_stream::{Stream, StreamExt}; /// /// use std::pin::Pin; /// use std::task::{Context, Poll}; @@ -498,7 +540,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// let mut stream = stream::iter(1..=10).take(3); /// @@ -527,7 +569,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// let mut stream = stream::iter(1..=10).take_while(|x| *x <= 3); /// @@ -553,7 +595,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// let mut stream = stream::iter(1..=10).skip(7); /// @@ -584,7 +626,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// let mut stream = stream::iter(vec![1,2,3,4,1]).skip_while(|x| *x < 3); /// /// assert_eq!(Some(3), stream.next().await); @@ -627,7 +669,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// let a = [1, 2, 3]; /// @@ -642,7 +684,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// let a = [1, 2, 3]; /// @@ -686,7 +728,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// let a = [1, 2, 3]; /// @@ -701,7 +743,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// let a = [1, 2, 3]; /// @@ -730,7 +772,7 @@ pub trait StreamExt: Stream { /// # Examples /// /// ``` - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// #[tokio::main] /// async fn main() { @@ -770,7 +812,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, *}; + /// use tokio_stream::{self as stream, *}; /// /// let s = stream::iter(vec![1u8, 2, 3]); /// let sum = s.fold(0, |acc, x| acc + x).await; @@ -797,7 +839,9 @@ pub trait StreamExt: Stream { /// `collect` streams all values, awaiting as needed. Values are pushed into /// a collection. A number of different target collection types are /// supported, including [`Vec`](std::vec::Vec), - /// [`String`](std::string::String), and [`Bytes`](bytes::Bytes). + /// [`String`](std::string::String), and [`Bytes`]. + /// + /// [`Bytes`]: https://docs.rs/bytes/0.6.0/bytes/struct.Bytes.html /// /// # `Result` /// @@ -816,7 +860,7 @@ pub trait StreamExt: Stream { /// Basic usage: /// /// ``` - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// #[tokio::main] /// async fn main() { @@ -833,7 +877,7 @@ pub trait StreamExt: Stream { /// Collecting a stream of `Result` values /// /// ``` - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// /// #[tokio::main] /// async fn main() { @@ -889,7 +933,7 @@ pub trait StreamExt: Stream { /// ``` /// # #[tokio::main] /// # async fn main() { - /// use tokio::stream::{self, StreamExt}; + /// use tokio_stream::{self as stream, StreamExt}; /// use std::time::Duration; /// # let int_stream = stream::iter(1..=3); /// @@ -934,7 +978,7 @@ pub trait StreamExt: Stream { /// Create a throttled stream. /// ```rust,no_run /// use std::time::Duration; - /// use tokio::stream::StreamExt; + /// use tokio_stream::StreamExt; /// /// # async fn dox() { /// let mut item_stream = futures::stream::repeat("one").throttle(Duration::from_secs(2)); diff --git a/tokio-stream/src/macros.rs b/tokio-stream/src/macros.rs new file mode 100644 index 000000000..0d493a851 --- /dev/null +++ b/tokio-stream/src/macros.rs @@ -0,0 +1,18 @@ +macro_rules! cfg_time { + ($($item:item)*) => { + $( + #[cfg(feature = "time")] + #[cfg_attr(docsrs, doc(cfg(feature = "time")))] + $item + )* + } +} + +macro_rules! ready { + ($e:expr $(,)?) => { + match $e { + std::task::Poll::Ready(t) => t, + std::task::Poll::Pending => return std::task::Poll::Pending, + } + }; +} diff --git a/tokio/src/stream/map.rs b/tokio-stream/src/map.rs similarity index 97% rename from tokio/src/stream/map.rs rename to tokio-stream/src/map.rs index dfac5a2c9..e6b47cd25 100644 --- a/tokio/src/stream/map.rs +++ b/tokio-stream/src/map.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::fmt; use core::pin::Pin; diff --git a/tokio/src/stream/merge.rs b/tokio-stream/src/merge.rs similarity index 98% rename from tokio/src/stream/merge.rs rename to tokio-stream/src/merge.rs index 50ba518ce..ea0ace0e2 100644 --- a/tokio/src/stream/merge.rs +++ b/tokio-stream/src/merge.rs @@ -1,4 +1,4 @@ -use crate::stream::{Fuse, Stream}; +use crate::{Fuse, Stream}; use core::pin::Pin; use core::task::{Context, Poll}; diff --git a/tokio/src/stream/next.rs b/tokio-stream/src/next.rs similarity index 97% rename from tokio/src/stream/next.rs rename to tokio-stream/src/next.rs index d9b1f9205..175490c48 100644 --- a/tokio/src/stream/next.rs +++ b/tokio-stream/src/next.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::future::Future; use core::marker::PhantomPinned; diff --git a/tokio/src/stream/once.rs b/tokio-stream/src/once.rs similarity index 88% rename from tokio/src/stream/once.rs rename to tokio-stream/src/once.rs index 7fe204cc1..04b4c052b 100644 --- a/tokio/src/stream/once.rs +++ b/tokio-stream/src/once.rs @@ -1,4 +1,4 @@ -use crate::stream::{self, Iter, Stream}; +use crate::{Iter, Stream}; use core::option; use core::pin::Pin; @@ -20,7 +20,7 @@ impl Unpin for Once {} /// # Examples /// /// ``` -/// use tokio::stream::{self, StreamExt}; +/// use tokio_stream::{self as stream, StreamExt}; /// /// #[tokio::main] /// async fn main() { @@ -35,7 +35,7 @@ impl Unpin for Once {} /// ``` pub fn once(value: T) -> Once { Once { - iter: stream::iter(Some(value).into_iter()), + iter: crate::iter(Some(value).into_iter()), } } diff --git a/tokio/src/stream/pending.rs b/tokio-stream/src/pending.rs similarity index 89% rename from tokio/src/stream/pending.rs rename to tokio-stream/src/pending.rs index 21224c385..b50fd3335 100644 --- a/tokio/src/stream/pending.rs +++ b/tokio-stream/src/pending.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::marker::PhantomData; use core::pin::Pin; @@ -16,7 +16,7 @@ unsafe impl Sync for Pending {} /// Creates a stream that is never ready /// /// The returned stream is never ready. Attempting to call -/// [`next()`](crate::stream::StreamExt::next) will never complete. Use +/// [`next()`](crate::StreamExt::next) will never complete. Use /// [`stream::empty()`](super::empty()) to obtain a stream that is is /// immediately empty but returns no values. /// @@ -25,7 +25,7 @@ unsafe impl Sync for Pending {} /// Basic usage: /// /// ```no_run -/// use tokio::stream::{self, StreamExt}; +/// use tokio_stream::{self as stream, StreamExt}; /// /// #[tokio::main] /// async fn main() { diff --git a/tokio/src/stream/skip.rs b/tokio-stream/src/skip.rs similarity index 98% rename from tokio/src/stream/skip.rs rename to tokio-stream/src/skip.rs index 39540cc98..80a0a0aff 100644 --- a/tokio/src/stream/skip.rs +++ b/tokio-stream/src/skip.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::fmt; use core::pin::Pin; diff --git a/tokio/src/stream/skip_while.rs b/tokio-stream/src/skip_while.rs similarity index 98% rename from tokio/src/stream/skip_while.rs rename to tokio-stream/src/skip_while.rs index 4e0500701..985a92666 100644 --- a/tokio/src/stream/skip_while.rs +++ b/tokio-stream/src/skip_while.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::fmt; use core::pin::Pin; diff --git a/tokio/src/stream/stream_map.rs b/tokio-stream/src/stream_map.rs similarity index 77% rename from tokio/src/stream/stream_map.rs rename to tokio-stream/src/stream_map.rs index 9fed3c196..85b60cfea 100644 --- a/tokio/src/stream/stream_map.rs +++ b/tokio-stream/src/stream_map.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use std::borrow::Borrow; use std::hash::Hash; @@ -42,9 +42,9 @@ use std::task::{Context, Poll}; /// to be merged, it may be advisable to use tasks sending values on a shared /// [`mpsc`] channel. /// -/// [`StreamExt::merge`]: crate::stream::StreamExt::merge -/// [`mpsc`]: crate::sync::mpsc -/// [`pin!`]: macro@pin +/// [`StreamExt::merge`]: crate::StreamExt::merge +/// [`mpsc`]: https://docs.rs/tokio/1.0/tokio/sync/mpsc/index.html +/// [`pin!`]: https://docs.rs/tokio/1.0/tokio/macro.pin.html /// [`Box::pin`]: std::boxed::Box::pin /// /// # Examples @@ -52,13 +52,27 @@ use std::task::{Context, Poll}; /// Merging two streams, then remove them after receiving the first value /// /// ``` -/// use tokio::stream::{StreamExt, StreamMap}; +/// use tokio_stream::{StreamExt, StreamMap, Stream}; /// use tokio::sync::mpsc; +/// use std::pin::Pin; /// /// #[tokio::main] /// async fn main() { -/// let (tx1, rx1) = mpsc::channel(10); -/// let (tx2, rx2) = mpsc::channel(10); +/// let (tx1, mut rx1) = mpsc::channel::(10); +/// let (tx2, mut rx2) = mpsc::channel::(10); +/// +/// // Convert the channels to a `Stream`. +/// let rx1 = Box::pin(async_stream::stream! { +/// while let Some(item) = rx1.recv().await { +/// yield item; +/// } +/// }) as Pin + Send>>; +/// +/// let rx2 = Box::pin(async_stream::stream! { +/// while let Some(item) = rx2.recv().await { +/// yield item; +/// } +/// }) as Pin + Send>>; /// /// tokio::spawn(async move { /// tx1.send(1).await.unwrap(); @@ -103,7 +117,7 @@ use std::task::{Context, Poll}; /// sent to the client over a socket. /// /// ```no_run -/// use tokio::stream::{Stream, StreamExt, StreamMap}; +/// use tokio_stream::{Stream, StreamExt, StreamMap}; /// /// enum Command { /// Join(String), @@ -112,13 +126,13 @@ use std::task::{Context, Poll}; /// /// fn commands() -> impl Stream { /// // Streams in user commands by parsing `stdin`. -/// # tokio::stream::pending() +/// # tokio_stream::pending() /// } /// /// // Join a channel, returns a stream of messages received on the channel. /// fn join(channel: &str) -> impl Stream + Unpin { /// // left as an exercise to the reader -/// # tokio::stream::pending() +/// # tokio_stream::pending() /// } /// /// #[tokio::main] @@ -170,7 +184,7 @@ impl StreamMap { /// # Examples /// /// ``` - /// use tokio::stream::{StreamMap, pending}; + /// use tokio_stream::{StreamMap, pending}; /// /// let mut map = StreamMap::new(); /// @@ -193,7 +207,7 @@ impl StreamMap { /// # Examples /// /// ``` - /// use tokio::stream::{StreamMap, pending}; + /// use tokio_stream::{StreamMap, pending}; /// /// let mut map = StreamMap::new(); /// @@ -217,7 +231,7 @@ impl StreamMap { /// # Examples /// /// ``` - /// use tokio::stream::{StreamMap, Pending}; + /// use tokio_stream::{StreamMap, Pending}; /// /// let map: StreamMap<&str, Pending<()>> = StreamMap::new(); /// ``` @@ -233,7 +247,7 @@ impl StreamMap { /// # Examples /// /// ``` - /// use tokio::stream::{StreamMap, Pending}; + /// use tokio_stream::{StreamMap, Pending}; /// /// let map: StreamMap<&str, Pending<()>> = StreamMap::with_capacity(10); /// ``` @@ -250,7 +264,7 @@ impl StreamMap { /// # Examples /// /// ``` - /// use tokio::stream::{StreamMap, pending}; + /// use tokio_stream::{StreamMap, pending}; /// /// let mut map = StreamMap::new(); /// @@ -273,7 +287,7 @@ impl StreamMap { /// # Examples /// /// ``` - /// use tokio::stream::{StreamMap, pending}; + /// use tokio_stream::{StreamMap, pending}; /// /// let mut map = StreamMap::new(); /// @@ -296,7 +310,7 @@ impl StreamMap { /// # Examples /// /// ``` - /// use tokio::stream::{StreamMap, pending}; + /// use tokio_stream::{StreamMap, pending}; /// /// let mut map = StreamMap::new(); /// @@ -320,7 +334,7 @@ impl StreamMap { /// # Examples /// /// ``` - /// use tokio::stream::{StreamMap, Pending}; + /// use tokio_stream::{StreamMap, Pending}; /// /// let map: StreamMap> = StreamMap::with_capacity(100); /// assert!(map.capacity() >= 100); @@ -334,7 +348,7 @@ impl StreamMap { /// # Examples /// /// ``` - /// use tokio::stream::{StreamMap, pending}; + /// use tokio_stream::{StreamMap, pending}; /// /// let mut a = StreamMap::new(); /// assert_eq!(a.len(), 0); @@ -367,7 +381,7 @@ impl StreamMap { /// # Examples /// /// ``` - /// use tokio::stream::{StreamMap, pending}; + /// use tokio_stream::{StreamMap, pending}; /// /// let mut a = StreamMap::new(); /// a.insert(1, pending::()); @@ -388,7 +402,7 @@ impl StreamMap { /// # Examples /// /// ``` - /// use tokio::stream::{StreamMap, pending}; + /// use tokio_stream::{StreamMap, pending}; /// /// let mut map = StreamMap::new(); /// @@ -416,7 +430,7 @@ impl StreamMap { /// # Examples /// /// ``` - /// use tokio::stream::{StreamMap, pending}; + /// use tokio_stream::{StreamMap, pending}; /// /// let mut map = StreamMap::new(); /// map.insert(1, pending::()); @@ -445,7 +459,7 @@ impl StreamMap { /// # Examples /// /// ``` - /// use tokio::stream::{StreamMap, pending}; + /// use tokio_stream::{StreamMap, pending}; /// /// let mut map = StreamMap::new(); /// map.insert(1, pending::()); @@ -476,7 +490,7 @@ where fn poll_next_entry(&mut self, cx: &mut Context<'_>) -> Poll> { use Poll::*; - let start = crate::util::thread_rng_n(self.entries.len() as u32) as usize; + let start = self::rand::thread_rng_n(self.entries.len() as u32) as usize; let mut idx = start; for _ in 0..self.entries.len() { @@ -553,3 +567,98 @@ where ret } } + +mod rand { + use std::cell::Cell; + + mod loom { + #[cfg(not(loom))] + pub(crate) mod rand { + use std::collections::hash_map::RandomState; + use std::hash::{BuildHasher, Hash, Hasher}; + use std::sync::atomic::AtomicU32; + use std::sync::atomic::Ordering::Relaxed; + + static COUNTER: AtomicU32 = AtomicU32::new(1); + + pub(crate) fn seed() -> u64 { + let rand_state = RandomState::new(); + + let mut hasher = rand_state.build_hasher(); + + // Hash some unique-ish data to generate some new state + COUNTER.fetch_add(1, Relaxed).hash(&mut hasher); + + // Get the seed + hasher.finish() + } + } + + #[cfg(loom)] + pub(crate) mod rand { + pub(crate) fn seed() -> u64 { + 1 + } + } + } + + /// Fast random number generate + /// + /// Implement xorshift64+: 2 32-bit xorshift sequences added together. + /// Shift triplet [17,7,16] was calculated as indicated in Marsaglia's + /// Xorshift paper: https://www.jstatsoft.org/article/view/v008i14/xorshift.pdf + /// This generator passes the SmallCrush suite, part of TestU01 framework: + /// http://simul.iro.umontreal.ca/testu01/tu01.html + #[derive(Debug)] + pub(crate) struct FastRand { + one: Cell, + two: Cell, + } + + impl FastRand { + /// Initialize a new, thread-local, fast random number generator. + pub(crate) fn new(seed: u64) -> FastRand { + let one = (seed >> 32) as u32; + let mut two = seed as u32; + + if two == 0 { + // This value cannot be zero + two = 1; + } + + FastRand { + one: Cell::new(one), + two: Cell::new(two), + } + } + + pub(crate) fn fastrand_n(&self, n: u32) -> u32 { + // This is similar to fastrand() % n, but faster. + // See https://lemire.me/blog/2016/06/27/a-fast-alternative-to-the-modulo-reduction/ + let mul = (self.fastrand() as u64).wrapping_mul(n as u64); + (mul >> 32) as u32 + } + + fn fastrand(&self) -> u32 { + let mut s1 = self.one.get(); + let s0 = self.two.get(); + + s1 ^= s1 << 17; + s1 = s1 ^ s0 ^ s1 >> 7 ^ s0 >> 16; + + self.one.set(s0); + self.two.set(s1); + + s0.wrapping_add(s1) + } + } + + // Used by `StreamMap` + pub(crate) fn thread_rng_n(n: u32) -> u32 { + thread_local! { + static THREAD_RNG: FastRand = FastRand::new(loom::rand::seed()); + } + + THREAD_RNG.with(|rng| rng.fastrand_n(n)) + } +} diff --git a/tokio/src/stream/take.rs b/tokio-stream/src/take.rs similarity index 98% rename from tokio/src/stream/take.rs rename to tokio-stream/src/take.rs index a92430b77..c75648f60 100644 --- a/tokio/src/stream/take.rs +++ b/tokio-stream/src/take.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::cmp; use core::fmt; diff --git a/tokio/src/stream/take_while.rs b/tokio-stream/src/take_while.rs similarity index 98% rename from tokio/src/stream/take_while.rs rename to tokio-stream/src/take_while.rs index cf1e16061..5ce4dd98a 100644 --- a/tokio/src/stream/take_while.rs +++ b/tokio-stream/src/take_while.rs @@ -1,4 +1,4 @@ -use crate::stream::Stream; +use crate::Stream; use core::fmt; use core::pin::Pin; diff --git a/tokio/src/stream/throttle.rs b/tokio-stream/src/throttle.rs similarity index 95% rename from tokio/src/stream/throttle.rs rename to tokio-stream/src/throttle.rs index ff1fbf01f..525763dca 100644 --- a/tokio/src/stream/throttle.rs +++ b/tokio-stream/src/throttle.rs @@ -1,7 +1,7 @@ //! Slow down a stream by enforcing a delay between items. -use crate::stream::Stream; -use crate::time::{Duration, Instant, Sleep}; +use crate::Stream; +use tokio::time::{Duration, Instant, Sleep}; use std::future::Future; use std::marker::Unpin; @@ -17,7 +17,7 @@ where let delay = if duration == Duration::from_millis(0) { None } else { - Some(Sleep::new_timeout(Instant::now() + duration)) + Some(tokio::time::sleep_until(Instant::now() + duration)) }; Throttle { diff --git a/tokio/src/stream/timeout.rs b/tokio-stream/src/timeout.rs similarity index 72% rename from tokio/src/stream/timeout.rs rename to tokio-stream/src/timeout.rs index 61154da05..303142f01 100644 --- a/tokio/src/stream/timeout.rs +++ b/tokio-stream/src/timeout.rs @@ -1,10 +1,11 @@ -use crate::stream::{Fuse, Stream}; -use crate::time::{error::Elapsed, Instant, Sleep}; +use crate::{Fuse, 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! { @@ -20,10 +21,14 @@ pin_project! { } } +/// Error returned by `Timeout`. +#[derive(Debug, PartialEq)] +pub struct Elapsed(()); + impl Timeout { pub(super) fn new(stream: S, duration: Duration) -> Self { let next = Instant::now() + duration; - let deadline = Sleep::new_timeout(next); + let deadline = tokio::time::sleep_until(next); Timeout { stream: Fuse::new(stream), @@ -63,3 +68,25 @@ impl Stream for Timeout { self.stream.size_hint() } } + +// ===== 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 for std::io::Error { + fn from(_err: Elapsed) -> std::io::Error { + std::io::ErrorKind::TimedOut.into() + } +} diff --git a/tokio/src/stream/try_next.rs b/tokio-stream/src/try_next.rs similarity index 96% rename from tokio/src/stream/try_next.rs rename to tokio-stream/src/try_next.rs index b21d279af..e91e8a429 100644 --- a/tokio/src/stream/try_next.rs +++ b/tokio-stream/src/try_next.rs @@ -1,4 +1,4 @@ -use crate::stream::{Next, Stream}; +use crate::{Next, Stream}; use core::future::Future; use core::marker::PhantomPinned; diff --git a/tokio-stream/tests/async_send_sync.rs b/tokio-stream/tests/async_send_sync.rs new file mode 100644 index 000000000..c06bebd22 --- /dev/null +++ b/tokio-stream/tests/async_send_sync.rs @@ -0,0 +1,105 @@ +use std::rc::Rc; + +#[allow(dead_code)] +type BoxStream = std::pin::Pin>>; + +#[allow(dead_code)] +fn require_send(_t: &T) {} +#[allow(dead_code)] +fn require_sync(_t: &T) {} +#[allow(dead_code)] +fn require_unpin(_t: &T) {} + +#[allow(dead_code)] +struct Invalid; + +trait AmbiguousIfSend { + fn some_item(&self) {} +} +impl AmbiguousIfSend<()> for T {} +impl AmbiguousIfSend for T {} + +trait AmbiguousIfSync { + fn some_item(&self) {} +} +impl AmbiguousIfSync<()> for T {} +impl AmbiguousIfSync for T {} + +trait AmbiguousIfUnpin { + fn some_item(&self) {} +} +impl AmbiguousIfUnpin<()> for T {} +impl AmbiguousIfUnpin 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>(): Send & Sync); +async_assert_fn!(tokio_stream::pending>(): Send & Sync); +async_assert_fn!(tokio_stream::iter(std::vec::IntoIter): Send & Sync); + +async_assert_fn!(tokio_stream::StreamExt::next(&mut BoxStream<()>): !Unpin); +async_assert_fn!(tokio_stream::StreamExt::try_next(&mut BoxStream>): !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>(&mut BoxStream<()>): !Unpin); diff --git a/tokio/tests/stream_chain.rs b/tokio-stream/tests/stream_chain.rs similarity index 89% rename from tokio/tests/stream_chain.rs rename to tokio-stream/tests/stream_chain.rs index 98461a8cc..759de3068 100644 --- a/tokio/tests/stream_chain.rs +++ b/tokio-stream/tests/stream_chain.rs @@ -1,7 +1,12 @@ -use tokio::stream::{self, Stream, StreamExt}; -use tokio::sync::mpsc; +use tokio_stream::{self as stream, Stream, StreamExt}; use tokio_test::{assert_pending, assert_ready, task}; +mod support { + pub(crate) mod mpsc; +} + +use support::mpsc; + #[tokio::test] async fn basic_usage() { let one = stream::iter(vec![1, 2, 3]); @@ -36,8 +41,8 @@ async fn basic_usage() { #[tokio::test] async fn pending_first() { - let (tx1, rx1) = mpsc::unbounded_channel(); - let (tx2, rx2) = mpsc::unbounded_channel(); + let (tx1, rx1) = mpsc::unbounded_channel_stream(); + let (tx2, rx2) = mpsc::unbounded_channel_stream(); let mut stream = task::spawn(rx1.chain(rx2)); assert_eq!(stream.size_hint(), (0, None)); @@ -74,7 +79,7 @@ async fn pending_first() { fn size_overflow() { struct Monster; - impl tokio::stream::Stream for Monster { + impl tokio_stream::Stream for Monster { type Item = (); fn poll_next( self: std::pin::Pin<&mut Self>, diff --git a/tokio/tests/stream_collect.rs b/tokio-stream/tests/stream_collect.rs similarity index 89% rename from tokio/tests/stream_collect.rs rename to tokio-stream/tests/stream_collect.rs index 7ab1a34ed..07659a1fc 100644 --- a/tokio/tests/stream_collect.rs +++ b/tokio-stream/tests/stream_collect.rs @@ -1,7 +1,12 @@ -use tokio::stream::{self, StreamExt}; -use tokio::sync::mpsc; +use tokio_stream::{self as stream, StreamExt}; use tokio_test::{assert_pending, assert_ready, assert_ready_err, assert_ready_ok, task}; +mod support { + pub(crate) mod mpsc; +} + +use support::mpsc; + #[allow(clippy::let_unit_value)] #[tokio::test] async fn empty_unit() { @@ -37,7 +42,7 @@ async fn empty_result() { #[tokio::test] async fn collect_vec_items() { - let (tx, rx) = mpsc::unbounded_channel(); + let (tx, rx) = mpsc::unbounded_channel_stream(); let mut fut = task::spawn(rx.collect::>()); assert_pending!(fut.poll()); @@ -58,7 +63,8 @@ async fn collect_vec_items() { #[tokio::test] async fn collect_string_items() { - let (tx, rx) = mpsc::unbounded_channel(); + let (tx, rx) = mpsc::unbounded_channel_stream(); + let mut fut = task::spawn(rx.collect::()); assert_pending!(fut.poll()); @@ -79,7 +85,8 @@ async fn collect_string_items() { #[tokio::test] async fn collect_str_items() { - let (tx, rx) = mpsc::unbounded_channel(); + let (tx, rx) = mpsc::unbounded_channel_stream(); + let mut fut = task::spawn(rx.collect::()); assert_pending!(fut.poll()); @@ -100,7 +107,8 @@ async fn collect_str_items() { #[tokio::test] async fn collect_results_ok() { - let (tx, rx) = mpsc::unbounded_channel(); + let (tx, rx) = mpsc::unbounded_channel_stream(); + let mut fut = task::spawn(rx.collect::>()); assert_pending!(fut.poll()); @@ -121,7 +129,8 @@ async fn collect_results_ok() { #[tokio::test] async fn collect_results_err() { - let (tx, rx) = mpsc::unbounded_channel(); + let (tx, rx) = mpsc::unbounded_channel_stream(); + let mut fut = task::spawn(rx.collect::>()); assert_pending!(fut.poll()); diff --git a/tokio/tests/stream_empty.rs b/tokio-stream/tests/stream_empty.rs similarity index 79% rename from tokio/tests/stream_empty.rs rename to tokio-stream/tests/stream_empty.rs index f278076d1..c06f5c41c 100644 --- a/tokio/tests/stream_empty.rs +++ b/tokio-stream/tests/stream_empty.rs @@ -1,4 +1,4 @@ -use tokio::stream::{self, Stream, StreamExt}; +use tokio_stream::{self as stream, Stream, StreamExt}; #[tokio::test] async fn basic_usage() { diff --git a/tokio/tests/stream_fuse.rs b/tokio-stream/tests/stream_fuse.rs similarity index 96% rename from tokio/tests/stream_fuse.rs rename to tokio-stream/tests/stream_fuse.rs index 9d7d969f8..9b6cf054c 100644 --- a/tokio/tests/stream_fuse.rs +++ b/tokio-stream/tests/stream_fuse.rs @@ -1,4 +1,4 @@ -use tokio::stream::{Stream, StreamExt}; +use tokio_stream::{Stream, StreamExt}; use std::pin::Pin; use std::task::{Context, Poll}; diff --git a/tokio/tests/stream_iter.rs b/tokio-stream/tests/stream_iter.rs similarity index 91% rename from tokio/tests/stream_iter.rs rename to tokio-stream/tests/stream_iter.rs index 45148a7a8..8b9ee3ce5 100644 --- a/tokio/tests/stream_iter.rs +++ b/tokio-stream/tests/stream_iter.rs @@ -1,4 +1,4 @@ -use tokio::stream; +use tokio_stream as stream; use tokio_test::task; use std::iter; diff --git a/tokio/tests/stream_merge.rs b/tokio-stream/tests/stream_merge.rs similarity index 86% rename from tokio/tests/stream_merge.rs rename to tokio-stream/tests/stream_merge.rs index 45ecdcb66..69cd568a6 100644 --- a/tokio/tests/stream_merge.rs +++ b/tokio-stream/tests/stream_merge.rs @@ -1,8 +1,13 @@ -use tokio::stream::{self, Stream, StreamExt}; -use tokio::sync::mpsc; +use tokio_stream::{self as stream, Stream, StreamExt}; use tokio_test::task; use tokio_test::{assert_pending, assert_ready}; +mod support { + pub(crate) mod mpsc; +} + +use support::mpsc; + #[tokio::test] async fn merge_sync_streams() { let mut s = stream::iter(vec![0, 2, 4, 6]).merge(stream::iter(vec![1, 3, 5])); @@ -18,8 +23,8 @@ async fn merge_sync_streams() { #[tokio::test] async fn merge_async_streams() { - let (tx1, rx1) = mpsc::unbounded_channel(); - let (tx2, rx2) = mpsc::unbounded_channel(); + let (tx1, rx1) = mpsc::unbounded_channel_stream(); + let (tx2, rx2) = mpsc::unbounded_channel_stream(); let mut rx = task::spawn(rx1.merge(rx2)); @@ -57,7 +62,7 @@ async fn merge_async_streams() { fn size_overflow() { struct Monster; - impl tokio::stream::Stream for Monster { + impl tokio_stream::Stream for Monster { type Item = (); fn poll_next( self: std::pin::Pin<&mut Self>, diff --git a/tokio/tests/stream_once.rs b/tokio-stream/tests/stream_once.rs similarity index 82% rename from tokio/tests/stream_once.rs rename to tokio-stream/tests/stream_once.rs index bb4635ac9..f32bad3a1 100644 --- a/tokio/tests/stream_once.rs +++ b/tokio-stream/tests/stream_once.rs @@ -1,4 +1,4 @@ -use tokio::stream::{self, Stream, StreamExt}; +use tokio_stream::{self as stream, Stream, StreamExt}; #[tokio::test] async fn basic_usage() { diff --git a/tokio/tests/stream_pending.rs b/tokio-stream/tests/stream_pending.rs similarity index 85% rename from tokio/tests/stream_pending.rs rename to tokio-stream/tests/stream_pending.rs index f4d3080de..87b5d03bd 100644 --- a/tokio/tests/stream_pending.rs +++ b/tokio-stream/tests/stream_pending.rs @@ -1,4 +1,4 @@ -use tokio::stream::{self, Stream, StreamExt}; +use tokio_stream::{self as stream, Stream, StreamExt}; use tokio_test::{assert_pending, task}; #[tokio::test] diff --git a/tokio/tests/stream_stream_map.rs b/tokio-stream/tests/stream_stream_map.rs similarity index 93% rename from tokio/tests/stream_stream_map.rs rename to tokio-stream/tests/stream_stream_map.rs index 38bb0c5d0..53f3d86c7 100644 --- a/tokio/tests/stream_stream_map.rs +++ b/tokio-stream/tests/stream_stream_map.rs @@ -1,7 +1,12 @@ -use tokio::stream::{self, pending, Stream, StreamExt, StreamMap}; -use tokio::sync::mpsc; +use tokio_stream::{self as stream, pending, Stream, StreamExt, StreamMap}; use tokio_test::{assert_ok, assert_pending, assert_ready, task}; +mod support { + pub(crate) mod mpsc; +} + +use support::mpsc; + use std::pin::Pin; macro_rules! assert_ready_some { @@ -38,7 +43,8 @@ async fn empty() { #[tokio::test] async fn single_entry() { let mut map = task::spawn(StreamMap::new()); - let (tx, rx) = mpsc::unbounded_channel(); + let (tx, rx) = mpsc::unbounded_channel_stream(); + let rx = Box::pin(rx); assert_ready_none!(map.poll_next()); @@ -76,8 +82,11 @@ async fn single_entry() { #[tokio::test] async fn multiple_entries() { let mut map = task::spawn(StreamMap::new()); - let (tx1, rx1) = mpsc::unbounded_channel(); - let (tx2, rx2) = mpsc::unbounded_channel(); + let (tx1, rx1) = mpsc::unbounded_channel_stream(); + let (tx2, rx2) = mpsc::unbounded_channel_stream(); + + let rx1 = Box::pin(rx1); + let rx2 = Box::pin(rx2); map.insert("foo", rx1); map.insert("bar", rx2); @@ -132,7 +141,9 @@ async fn multiple_entries() { #[tokio::test] async fn insert_remove() { let mut map = task::spawn(StreamMap::new()); - let (tx, rx) = mpsc::unbounded_channel(); + let (tx, rx) = mpsc::unbounded_channel_stream(); + + let rx = Box::pin(rx); assert_ready_none!(map.poll_next()); @@ -160,8 +171,11 @@ async fn insert_remove() { #[tokio::test] async fn replace() { let mut map = task::spawn(StreamMap::new()); - let (tx1, rx1) = mpsc::unbounded_channel(); - let (tx2, rx2) = mpsc::unbounded_channel(); + let (tx1, rx1) = mpsc::unbounded_channel_stream(); + let (tx2, rx2) = mpsc::unbounded_channel_stream(); + + let rx1 = Box::pin(rx1); + let rx2 = Box::pin(rx2); assert!(map.insert("foo", rx1).is_none()); diff --git a/tokio/tests/stream_timeout.rs b/tokio-stream/tests/stream_timeout.rs similarity index 98% rename from tokio/tests/stream_timeout.rs rename to tokio-stream/tests/stream_timeout.rs index 216b5f75d..5697ace69 100644 --- a/tokio/tests/stream_timeout.rs +++ b/tokio-stream/tests/stream_timeout.rs @@ -1,7 +1,7 @@ #![cfg(feature = "full")] -use tokio::stream::{self, StreamExt}; use tokio::time::{self, sleep, Duration}; +use tokio_stream::{self, StreamExt}; use tokio_test::*; use futures::StreamExt as _; diff --git a/tokio-stream/tests/support/mpsc.rs b/tokio-stream/tests/support/mpsc.rs new file mode 100644 index 000000000..09dbe0421 --- /dev/null +++ b/tokio-stream/tests/support/mpsc.rs @@ -0,0 +1,15 @@ +use async_stream::stream; +use tokio::sync::mpsc::{self, UnboundedSender}; +use tokio_stream::Stream; + +pub fn unbounded_channel_stream() -> (UnboundedSender, impl Stream) { + let (tx, mut rx) = mpsc::unbounded_channel(); + + let stream = stream! { + while let Some(item) = rx.recv().await { + yield item; + } + }; + + (tx, stream) +} diff --git a/tokio/tests/time_throttle.rs b/tokio-stream/tests/time_throttle.rs similarity index 95% rename from tokio/tests/time_throttle.rs rename to tokio-stream/tests/time_throttle.rs index c886319f6..42a643bfa 100644 --- a/tokio/tests/time_throttle.rs +++ b/tokio-stream/tests/time_throttle.rs @@ -1,8 +1,8 @@ #![warn(rust_2018_idioms)] #![cfg(feature = "full")] -use tokio::stream::StreamExt; use tokio::time; +use tokio_stream::StreamExt; use tokio_test::*; use std::time::Duration; diff --git a/tokio-test/Cargo.toml b/tokio-test/Cargo.toml index 79553c871..e6b8d661b 100644 --- a/tokio-test/Cargo.toml +++ b/tokio-test/Cargo.toml @@ -22,6 +22,8 @@ publish = false [dependencies] tokio = { version = "1.0.0", path = "../tokio", features = ["rt", "stream", "sync", "time", "test-util"] } +tokio-stream = { version = "0.1", path = "../tokio-stream" } +async-stream = "0.3" bytes = "0.6.0" futures-core = "0.3.0" diff --git a/tokio-test/src/io.rs b/tokio-test/src/io.rs index b68929a84..6705f80c3 100644 --- a/tokio-test/src/io.rs +++ b/tokio-test/src/io.rs @@ -22,8 +22,9 @@ use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; use tokio::sync::mpsc; use tokio::time::{self, Duration, Instant, Sleep}; -use futures_core::ready; +use futures_core::{ready, Stream}; use std::collections::VecDeque; +use std::fmt; use std::future::Future; use std::pin::Pin; use std::sync::Arc; @@ -63,13 +64,13 @@ enum Action { WriteError(Option>), } -#[derive(Debug)] struct Inner { actions: VecDeque, waiting: Option, sleep: Option, read_wait: Option, - rx: mpsc::UnboundedReceiver, + // rx: mpsc::UnboundedReceiver, + rx: Pin + Send>>, } impl Builder { @@ -184,7 +185,13 @@ impl Handle { impl Inner { fn new(actions: VecDeque) -> (Inner, Handle) { - let (tx, rx) = mpsc::unbounded_channel(); + let (tx, mut rx) = mpsc::unbounded_channel(); + + let rx = Box::pin(async_stream::stream! { + while let Some(item) = rx.recv().await { + yield item; + } + }); let inner = Inner { actions, @@ -200,8 +207,6 @@ impl Inner { } fn poll_action(&mut self, cx: &mut task::Context<'_>) -> Poll> { - use futures_core::stream::Stream; - Pin::new(&mut self.rx).poll_next(cx) } @@ -485,3 +490,9 @@ fn is_task_ctx() -> bool { r } */ + +impl fmt::Debug for Inner { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Inner {{...}}") + } +} diff --git a/tokio-test/src/task.rs b/tokio-test/src/task.rs index 6a73fd759..fa98bae0b 100644 --- a/tokio-test/src/task.rs +++ b/tokio-test/src/task.rs @@ -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(task: T) -> Spawn { diff --git a/tokio-util/Cargo.toml b/tokio-util/Cargo.toml index e567912ae..27b284135 100644 --- a/tokio-util/Cargo.toml +++ b/tokio-util/Cargo.toml @@ -38,6 +38,7 @@ __docs_rs = ["futures-util"] [dependencies] tokio = { version = "1.0.0", path = "../tokio" } +tokio-stream = { version = "0.1", path = "../tokio-stream" } bytes = "0.6.0" futures-core = "0.3.0" diff --git a/tokio-util/src/codec/decoder.rs b/tokio-util/src/codec/decoder.rs index 84d27fbf1..3d19332a2 100644 --- a/tokio-util/src/codec/decoder.rs +++ b/tokio-util/src/codec/decoder.rs @@ -153,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`]: tokio::stream::Stream + /// [`Stream`]: tokio_stream::Stream /// [`Sink`]: futures_sink::Sink /// [`Framed`]: crate::codec::Framed fn framed(self, io: T) -> Framed diff --git a/tokio-util/src/codec/framed.rs b/tokio-util/src/codec/framed.rs index 36370da26..adfe06380 100644 --- a/tokio-util/src/codec/framed.rs +++ b/tokio-util/src/codec/framed.rs @@ -2,10 +2,8 @@ use crate::codec::decoder::Decoder; use crate::codec::encoder::Encoder; use crate::codec::framed_impl::{FramedImpl, RWFrames, ReadFrame, WriteFrame}; -use tokio::{ - io::{AsyncRead, AsyncWrite}, - stream::Stream, -}; +use tokio::io::{AsyncRead, AsyncWrite}; +use tokio_stream::Stream; use bytes::BytesMut; use futures_sink::Sink; @@ -22,7 +20,7 @@ pin_project! { /// You can create a `Framed` instance by using the [`Decoder::framed`] adapter, or /// by using the `new` function seen below. /// - /// [`Stream`]: tokio::stream::Stream + /// [`Stream`]: tokio_stream::Stream /// [`Sink`]: futures_sink::Sink /// [`AsyncRead`]: tokio::io::AsyncRead /// [`Decoder::framed`]: crate::codec::Decoder::framed() @@ -54,7 +52,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`]: tokio::stream::Stream + /// [`Stream`]: tokio_stream::Stream /// [`Sink`]: futures_sink::Sink /// [`Decode`]: crate::codec::Decoder /// [`Encoder`]: crate::codec::Encoder @@ -88,7 +86,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`]: tokio::stream::Stream + /// [`Stream`]: tokio_stream::Stream /// [`Sink`]: futures_sink::Sink /// [`Decode`]: crate::codec::Decoder /// [`Encoder`]: crate::codec::Encoder @@ -133,7 +131,7 @@ impl Framed { /// calling [`split`] on the `Framed` returned by this method, which will /// break them into separate objects, allowing them to interact more easily. /// - /// [`Stream`]: tokio::stream::Stream + /// [`Stream`]: tokio_stream::Stream /// [`Sink`]: futures_sink::Sink /// [`Decoder`]: crate::codec::Decoder /// [`Encoder`]: crate::codec::Encoder diff --git a/tokio-util/src/codec/framed_impl.rs b/tokio-util/src/codec/framed_impl.rs index 207e198d2..69df6f2b1 100644 --- a/tokio-util/src/codec/framed_impl.rs +++ b/tokio-util/src/codec/framed_impl.rs @@ -1,10 +1,8 @@ use crate::codec::decoder::Decoder; use crate::codec::encoder::Encoder; -use tokio::{ - io::{AsyncRead, AsyncWrite}, - stream::Stream, -}; +use tokio::io::{AsyncRead, AsyncWrite}; +use tokio_stream::Stream; use bytes::BytesMut; use futures_core::ready; diff --git a/tokio-util/src/codec/framed_read.rs b/tokio-util/src/codec/framed_read.rs index 2077fbcee..d95589f23 100644 --- a/tokio-util/src/codec/framed_read.rs +++ b/tokio-util/src/codec/framed_read.rs @@ -1,7 +1,8 @@ use crate::codec::framed_impl::{FramedImpl, ReadFrame}; use crate::codec::Decoder; -use tokio::{io::AsyncRead, stream::Stream}; +use tokio::io::AsyncRead; +use tokio_stream::Stream; use bytes::BytesMut; use futures_sink::Sink; @@ -13,7 +14,7 @@ use std::task::{Context, Poll}; pin_project! { /// A [`Stream`] of messages decoded from an [`AsyncRead`]. /// - /// [`Stream`]: tokio::stream::Stream + /// [`Stream`]: tokio_stream::Stream /// [`AsyncRead`]: tokio::io::AsyncRead pub struct FramedRead { #[pin] diff --git a/tokio-util/src/codec/framed_write.rs b/tokio-util/src/codec/framed_write.rs index 834eb6ed0..8bb9efe41 100644 --- a/tokio-util/src/codec/framed_write.rs +++ b/tokio-util/src/codec/framed_write.rs @@ -1,7 +1,8 @@ use crate::codec::encoder::Encoder; use crate::codec::framed_impl::{FramedImpl, WriteFrame}; -use tokio::{io::AsyncWrite, stream::Stream}; +use tokio::io::AsyncWrite; +use tokio_stream::Stream; use futures_sink::Sink; use pin_project_lite::pin_project; diff --git a/tokio-util/src/codec/mod.rs b/tokio-util/src/codec/mod.rs index e89aa7c9a..12704eebf 100644 --- a/tokio-util/src/codec/mod.rs +++ b/tokio-util/src/codec/mod.rs @@ -11,7 +11,7 @@ //! //! [`AsyncRead`]: tokio::io::AsyncRead //! [`AsyncWrite`]: tokio::io::AsyncWrite -//! [`Stream`]: tokio::stream::Stream +//! [`Stream`]: tokio_stream::Stream //! [`Sink`]: futures_sink::Sink mod bytes_codec; diff --git a/tokio-util/src/either.rs b/tokio-util/src/either.rs index f5246af27..7f193de7a 100644 --- a/tokio-util/src/either.rs +++ b/tokio-util/src/either.rs @@ -167,10 +167,8 @@ where #[cfg(test)] mod tests { use super::*; - use tokio::{ - io::{repeat, AsyncReadExt, Repeat}, - stream::{once, Once, StreamExt}, - }; + use tokio::io::{repeat, AsyncReadExt, Repeat}; + use tokio_stream::{once, Once, StreamExt}; #[tokio::test] async fn either_is_stream() { diff --git a/tokio-util/src/io/read_buf.rs b/tokio-util/src/io/read_buf.rs index a5d46a7de..d7938a3bc 100644 --- a/tokio-util/src/io/read_buf.rs +++ b/tokio-util/src/io/read_buf.rs @@ -13,7 +13,7 @@ use tokio::io::AsyncRead; /// /// ``` /// use bytes::{Bytes, BytesMut}; -/// use tokio::stream; +/// use tokio_stream as stream; /// use tokio::io::Result; /// use tokio_util::io::{StreamReader, read_buf}; /// # #[tokio::main] diff --git a/tokio-util/src/io/reader_stream.rs b/tokio-util/src/io/reader_stream.rs index 3e6a05eff..7c25f3407 100644 --- a/tokio-util/src/io/reader_stream.rs +++ b/tokio-util/src/io/reader_stream.rs @@ -18,7 +18,7 @@ pin_project! { /// ``` /// # #[tokio::main] /// # async fn main() -> std::io::Result<()> { - /// use tokio::stream::StreamExt; + /// use tokio_stream::StreamExt; /// use tokio_util::io::ReaderStream; /// /// // Create a stream of data. @@ -40,7 +40,7 @@ pin_project! { /// /// [`AsyncRead`]: tokio::io::AsyncRead /// [`StreamReader`]: crate::io::StreamReader - /// [`Stream`]: tokio::stream::Stream + /// [`Stream`]: tokio_stream::Stream #[derive(Debug)] pub struct ReaderStream { // Reader itself. @@ -58,7 +58,7 @@ impl ReaderStream { /// `Result`. /// /// [`AsyncRead`]: tokio::io::AsyncRead - /// [`Stream`]: tokio::stream::Stream + /// [`Stream`]: tokio_stream::Stream pub fn new(reader: R) -> Self { ReaderStream { reader: Some(reader), diff --git a/tokio-util/src/io/stream_reader.rs b/tokio-util/src/io/stream_reader.rs index 99079c73d..12820de4a 100644 --- a/tokio-util/src/io/stream_reader.rs +++ b/tokio-util/src/io/stream_reader.rs @@ -21,7 +21,7 @@ pin_project! { /// # async fn main() -> std::io::Result<()> { /// /// // Create a stream from an iterator. - /// let stream = tokio::stream::iter(vec![ + /// let stream = tokio_stream::iter(vec![ /// Result::Ok(Bytes::from_static(&[0, 1, 2, 3])), /// Result::Ok(Bytes::from_static(&[4, 5, 6, 7])), /// Result::Ok(Bytes::from_static(&[8, 9, 10, 11])), @@ -51,7 +51,7 @@ pin_project! { /// ``` /// /// [`AsyncRead`]: tokio::io::AsyncRead - /// [`Stream`]: tokio::stream::Stream + /// [`Stream`]: tokio_stream::Stream /// [`ReaderStream`]: crate::io::ReaderStream #[derive(Debug)] pub struct StreamReader { diff --git a/tokio-util/src/lib.rs b/tokio-util/src/lib.rs index 4ae9a13da..999598c7a 100644 --- a/tokio-util/src/lib.rs +++ b/tokio-util/src/lib.rs @@ -72,7 +72,7 @@ mod util { /// /// ``` /// use bytes::{Bytes, BytesMut}; - /// use tokio::stream; + /// use tokio_stream as stream; /// use tokio::io::Result; /// use tokio_util::io::{StreamReader, poll_read_buf}; /// use futures::future::poll_fn; diff --git a/tokio-util/src/udp/frame.rs b/tokio-util/src/udp/frame.rs index 249acb706..b4ef3ca8c 100644 --- a/tokio-util/src/udp/frame.rs +++ b/tokio-util/src/udp/frame.rs @@ -1,6 +1,7 @@ use crate::codec::{Decoder, Encoder}; -use tokio::{io::ReadBuf, net::UdpSocket, stream::Stream}; +use tokio::{io::ReadBuf, net::UdpSocket}; +use tokio_stream::Stream; use bytes::{BufMut, BytesMut}; use futures_core::ready; @@ -27,7 +28,7 @@ use std::{io, mem::MaybeUninit}; /// calling [`split`] on the `UdpFramed` returned by this method, which will break /// them into separate objects, allowing them to interact more easily. /// -/// [`Stream`]: tokio::stream::Stream +/// [`Stream`]: tokio_stream::Stream /// [`Sink`]: futures_sink::Sink /// [`split`]: https://docs.rs/futures/0.3/futures/stream/trait.StreamExt.html#method.split #[must_use = "sinks do nothing unless polled"] diff --git a/tokio-util/tests/framed.rs b/tokio-util/tests/framed.rs index 4c5f84186..7e39e266a 100644 --- a/tokio-util/tests/framed.rs +++ b/tokio-util/tests/framed.rs @@ -1,6 +1,7 @@ #![warn(rust_2018_idioms)] -use tokio::{prelude::*, stream::StreamExt}; +use tokio::prelude::*; +use tokio_stream::StreamExt; use tokio_test::assert_ok; use tokio_util::codec::{Decoder, Encoder, Framed, FramedParts}; diff --git a/tokio-util/tests/io_reader_stream.rs b/tokio-util/tests/io_reader_stream.rs index 91986c8e3..e30cd8516 100644 --- a/tokio-util/tests/io_reader_stream.rs +++ b/tokio-util/tests/io_reader_stream.rs @@ -3,7 +3,7 @@ use std::pin::Pin; use std::task::{Context, Poll}; use tokio::io::{AsyncRead, ReadBuf}; -use tokio::stream::StreamExt; +use tokio_stream::StreamExt; /// produces at most `remaining` zeros, that returns error. /// each time it reads at most 31 byte. diff --git a/tokio-util/tests/io_stream_reader.rs b/tokio-util/tests/io_stream_reader.rs index b0ed1d2d0..59759941c 100644 --- a/tokio-util/tests/io_stream_reader.rs +++ b/tokio-util/tests/io_stream_reader.rs @@ -2,7 +2,7 @@ use bytes::Bytes; use tokio::io::AsyncReadExt; -use tokio::stream::iter; +use tokio_stream::iter; use tokio_util::io::StreamReader; #[tokio::test] diff --git a/tokio-util/tests/udp.rs b/tokio-util/tests/udp.rs index 4820ac72d..653d20deb 100644 --- a/tokio-util/tests/udp.rs +++ b/tokio-util/tests/udp.rs @@ -1,6 +1,7 @@ #![warn(rust_2018_idioms)] -use tokio::{net::UdpSocket, stream::StreamExt}; +use tokio::net::UdpSocket; +use tokio_stream::StreamExt; use tokio_util::codec::{Decoder, Encoder, LinesCodec}; use tokio_util::udp::UdpFramed; diff --git a/tokio/Cargo.toml b/tokio/Cargo.toml index 7c0f84b30..d565cc9d3 100644 --- a/tokio/Cargo.toml +++ b/tokio/Cargo.toml @@ -123,9 +123,11 @@ optional = true [dev-dependencies] tokio-test = { version = "0.4.0", path = "../tokio-test" } +tokio-stream = { version = "0.1", path = "../tokio-stream" } futures = { version = "0.3.0", features = ["async-await"] } proptest = "0.10.0" tempfile = "3.1.0" +async-stream = "0.3" [target.'cfg(loom)'.dev-dependencies] loom = { version = "0.3.5", features = ["futures", "checkpoint"] } diff --git a/tokio/src/coop.rs b/tokio/src/coop.rs index 980cdf8c4..05b2ae83b 100644 --- a/tokio/src/coop.rs +++ b/tokio/src/coop.rs @@ -13,7 +13,7 @@ //! Consider a future like this one: //! //! ``` -//! # use tokio::stream::{Stream, StreamExt}; +//! # use tokio_stream::{Stream, StreamExt}; //! async fn drop_all(mut input: I) { //! while let Some(_) = input.next().await {} //! } @@ -25,7 +25,7 @@ //! opt-in yield points, this problem is alleviated: //! //! ```ignore -//! # use tokio::stream::{Stream, StreamExt}; +//! # use tokio_stream::{Stream, StreamExt}; //! async fn drop_all(mut input: I) { //! while let Some(_) = input.next().await { //! tokio::coop::proceed().await; diff --git a/tokio/src/fs/read_dir.rs b/tokio/src/fs/read_dir.rs index 01d1ea832..219a0a305 100644 --- a/tokio/src/fs/read_dir.rs +++ b/tokio/src/fs/read_dir.rs @@ -29,12 +29,11 @@ pub async fn read_dir(path: impl AsRef) -> io::Result { /// /// # Errors /// -/// This [`Stream`] will return an [`Err`] if there's some sort of intermittent +/// This stream will return an [`Err`] if there's some sort of intermittent /// IO error during iteration. /// /// [`read_dir`]: read_dir /// [`DirEntry`]: DirEntry -/// [`Stream`]: crate::stream::Stream /// [`Err`]: std::result::Result::Err #[derive(Debug)] #[must_use = "streams do nothing unless polled"] @@ -111,19 +110,6 @@ feature! { } } -#[cfg(feature = "stream")] -impl crate::stream::Stream for ReadDir { - type Item = io::Result; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Poll::Ready(match ready!(self.poll_next_entry(cx)) { - Ok(Some(entry)) => Some(Ok(entry)), - Ok(None) => None, - Err(err) => Some(Err(err)), - }) - } -} - /// Entries returned by the [`ReadDir`] stream. /// /// [`ReadDir`]: struct@ReadDir diff --git a/tokio/src/io/mod.rs b/tokio/src/io/mod.rs index 14be3e06d..009f36ace 100644 --- a/tokio/src/io/mod.rs +++ b/tokio/src/io/mod.rs @@ -178,7 +178,7 @@ //! [`Read`]: std::io::Read //! [`SeekFrom`]: enum@SeekFrom //! [`Sink`]: https://docs.rs/futures/0.3/futures/sink/trait.Sink.html -//! [`Stream`]: crate::stream::Stream +//! [`Stream`]: https://docs.rs/futures/0.3/futures/stream/trait.Stream.html //! [`Write`]: std::io::Write cfg_io_blocking! { pub(crate) mod blocking; diff --git a/tokio/src/io/util/async_buf_read_ext.rs b/tokio/src/io/util/async_buf_read_ext.rs index 9e87f2f0d..7977a0e8c 100644 --- a/tokio/src/io/util/async_buf_read_ext.rs +++ b/tokio/src/io/util/async_buf_read_ext.rs @@ -228,7 +228,6 @@ cfg_io_util! { /// /// ``` /// use tokio::io::AsyncBufReadExt; - /// use tokio::stream::StreamExt; /// /// use std::io::Cursor; /// @@ -236,12 +235,12 @@ cfg_io_util! { /// async fn main() { /// let cursor = Cursor::new(b"lorem\nipsum\r\ndolor"); /// - /// let mut lines = cursor.lines().map(|res| res.unwrap()); + /// let mut lines = cursor.lines(); /// - /// assert_eq!(lines.next().await, Some(String::from("lorem"))); - /// assert_eq!(lines.next().await, Some(String::from("ipsum"))); - /// assert_eq!(lines.next().await, Some(String::from("dolor"))); - /// assert_eq!(lines.next().await, None); + /// assert_eq!(lines.next_line().await.unwrap(), Some(String::from("lorem"))); + /// assert_eq!(lines.next_line().await.unwrap(), Some(String::from("ipsum"))); + /// assert_eq!(lines.next_line().await.unwrap(), Some(String::from("dolor"))); + /// assert_eq!(lines.next_line().await.unwrap(), None); /// } /// ``` /// diff --git a/tokio/src/io/util/lines.rs b/tokio/src/io/util/lines.rs index b41f04a80..5ce249c0e 100644 --- a/tokio/src/io/util/lines.rs +++ b/tokio/src/io/util/lines.rs @@ -108,19 +108,6 @@ where } } -#[cfg(feature = "stream")] -impl crate::stream::Stream for Lines { - type Item = io::Result; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Poll::Ready(match ready!(self.poll_next_line(cx)) { - Ok(Some(line)) => Some(Ok(line)), - Ok(None) => None, - Err(err) => Some(Err(err)), - }) - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/tokio/src/io/util/split.rs b/tokio/src/io/util/split.rs index 492e26a63..75115aa08 100644 --- a/tokio/src/io/util/split.rs +++ b/tokio/src/io/util/split.rs @@ -89,19 +89,6 @@ where } } -#[cfg(feature = "stream")] -impl crate::stream::Stream for Split { - type Item = io::Result>; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Poll::Ready(match ready!(self.poll_next_segment(cx)) { - Ok(Some(segment)) => Some(Ok(segment)), - Ok(None) => None, - Err(err) => Some(Err(err)), - }) - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/tokio/src/lib.rs b/tokio/src/lib.rs index d5392b109..5cfefce76 100644 --- a/tokio/src/lib.rs +++ b/tokio/src/lib.rs @@ -389,10 +389,6 @@ cfg_signal_internal! { pub(crate) mod signal; } -cfg_stream! { - pub mod stream; -} - cfg_sync! { pub mod sync; } @@ -411,6 +407,13 @@ cfg_time! { mod util; +/// Due to the `Stream` trait's inclusion in `std` landing later than Tokio's 1.0 +/// release, most of the Tokio stream utilities have been moved into the [`tokio-stream`] +/// crate. +/// +/// [`tokio-stream`]: https://docs.rs/tokio-stream +pub mod stream {} + cfg_macros! { /// Implementation detail of the `select!` macro. This macro is **not** /// intended to be used as part of the public API and is permitted to diff --git a/tokio/src/macros/cfg.rs b/tokio/src/macros/cfg.rs index 49cd66fb9..9ae098fb0 100644 --- a/tokio/src/macros/cfg.rs +++ b/tokio/src/macros/cfg.rs @@ -241,16 +241,6 @@ macro_rules! cfg_not_signal_internal { } } -macro_rules! cfg_stream { - ($($item:item)*) => { - $( - #[cfg(feature = "stream")] - #[cfg_attr(docsrs, doc(cfg(feature = "stream")))] - $item - )* - } -} - macro_rules! cfg_sync { ($($item:item)*) => { $( @@ -361,7 +351,6 @@ macro_rules! cfg_coop { feature = "rt", feature = "signal", feature = "sync", - feature = "stream", feature = "time", ))] $item diff --git a/tokio/src/macros/pin.rs b/tokio/src/macros/pin.rs index ed844ef7d..a32187e5a 100644 --- a/tokio/src/macros/pin.rs +++ b/tokio/src/macros/pin.rs @@ -71,7 +71,7 @@ /// /// ``` /// use tokio::{pin, select}; -/// use tokio::stream::{self, StreamExt}; +/// use tokio_stream::{self as stream, StreamExt}; /// /// async fn my_async_fn() { /// // async logic here diff --git a/tokio/src/macros/select.rs b/tokio/src/macros/select.rs index b63abdd2d..2131c8907 100644 --- a/tokio/src/macros/select.rs +++ b/tokio/src/macros/select.rs @@ -167,7 +167,7 @@ /// Basic stream selecting. /// /// ``` -/// use tokio::stream::{self, StreamExt}; +/// use tokio_stream::{self as stream, StreamExt}; /// /// #[tokio::main] /// async fn main() { @@ -188,7 +188,7 @@ /// is complete, all calls to `next()` return `None`. /// /// ``` -/// use tokio::stream::{self, StreamExt}; +/// use tokio_stream::{self as stream, StreamExt}; /// /// #[tokio::main] /// async fn main() { @@ -220,7 +220,7 @@ /// Here, a stream is consumed for at most 1 second. /// /// ``` -/// use tokio::stream::{self, StreamExt}; +/// use tokio_stream::{self as stream, StreamExt}; /// use tokio::time::{self, Duration}; /// /// #[tokio::main] diff --git a/tokio/src/net/tcp/listener.rs b/tokio/src/net/tcp/listener.rs index 8b0a48036..9ac1b3e83 100644 --- a/tokio/src/net/tcp/listener.rs +++ b/tokio/src/net/tcp/listener.rs @@ -11,10 +11,8 @@ use std::task::{Context, Poll}; cfg_net! { /// A TCP socket server, listening for connections. /// - /// You can accept a new connection by using the [`accept`](`TcpListener::accept`) method. Alternatively `TcpListener` - /// implements the [`Stream`](`crate::stream::Stream`) trait, which allows you to use the listener in places that want a - /// stream. The stream will never return `None` and will also not yield the peer's `SocketAddr` structure. Iterating over - /// it is equivalent to calling accept in a loop. + /// You can accept a new connection by using the [`accept`](`TcpListener::accept`) + /// method. /// /// # Errors /// @@ -47,24 +45,6 @@ cfg_net! { /// } /// } /// ``` - /// - /// Using `impl Stream`: - /// ```no_run - /// use tokio::{net::TcpListener, stream::StreamExt}; - /// - /// #[tokio::main] - /// async fn main() { - /// let mut listener = TcpListener::bind("127.0.0.1:8080").await.unwrap(); - /// while let Some(stream) = listener.next().await { - /// match stream { - /// Ok(stream) => { - /// println!("new client!"); - /// } - /// Err(e) => { /* connection failed */ } - /// } - /// } - /// } - /// ``` pub struct TcpListener { io: PollEvented, } @@ -323,16 +303,6 @@ impl TcpListener { } } -#[cfg(feature = "stream")] -impl crate::stream::Stream for TcpListener { - type Item = io::Result; - - fn poll_next(self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let (socket, _) = ready!(self.poll_accept(cx))?; - Poll::Ready(Some(Ok(socket))) - } -} - impl TryFrom for TcpListener { type Error = io::Error; diff --git a/tokio/src/net/udp.rs b/tokio/src/net/udp.rs index 90210241e..23abe98e4 100644 --- a/tokio/src/net/udp.rs +++ b/tokio/src/net/udp.rs @@ -23,10 +23,11 @@ cfg_net! { /// /// # Streams /// - /// If you need to listen over UDP and produce a [`Stream`](`crate::stream::Stream`), you can look + /// If you need to listen over UDP and produce a [`Stream`], you can look /// at [`UdpFramed`]. /// /// [`UdpFramed`]: https://docs.rs/tokio-util/latest/tokio_util/udp/struct.UdpFramed.html + /// [`Stream`]: https://docs.rs/futures/0.3/futures/stream/trait.Stream.html /// /// # Example: one to many (bind) /// diff --git a/tokio/src/net/unix/listener.rs b/tokio/src/net/unix/listener.rs index b1da0e3c5..7d23a2e31 100644 --- a/tokio/src/net/unix/listener.rs +++ b/tokio/src/net/unix/listener.rs @@ -12,10 +12,7 @@ use std::task::{Context, Poll}; cfg_net_unix! { /// A Unix socket which can accept connections from other Unix sockets. /// - /// You can accept a new connection by using the [`accept`](`UnixListener::accept`) method. Alternatively `UnixListener` - /// implements the [`Stream`](`crate::stream::Stream`) trait, which allows you to use the listener in places that want a - /// stream. The stream will never return `None` and will also not yield the peer's `SocketAddr` structure. Iterating over - /// it is equivalent to calling accept in a loop. + /// You can accept a new connection by using the [`accept`](`UnixListener::accept`) method. /// /// # Errors /// @@ -29,14 +26,13 @@ cfg_net_unix! { /// /// ```no_run /// use tokio::net::UnixListener; - /// use tokio::stream::StreamExt; /// /// #[tokio::main] /// async fn main() { - /// let mut listener = UnixListener::bind("/path/to/the/socket").unwrap(); - /// while let Some(stream) = listener.next().await { - /// match stream { - /// Ok(stream) => { + /// let listener = UnixListener::bind("/path/to/the/socket").unwrap(); + /// loop { + /// match listener.accept().await { + /// Ok((stream, _addr)) => { /// println!("new client!"); /// } /// Err(e) => { /* connection failed */ } @@ -127,16 +123,6 @@ impl UnixListener { } } -#[cfg(feature = "stream")] -impl crate::stream::Stream for UnixListener { - type Item = io::Result; - - fn poll_next(self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let (socket, _) = ready!(self.poll_accept(cx))?; - Poll::Ready(Some(Ok(socket))) - } -} - impl TryFrom for UnixListener { type Error = io::Error; diff --git a/tokio/src/signal/unix.rs b/tokio/src/signal/unix.rs index aaaa75edb..fc0f16d4e 100644 --- a/tokio/src/signal/unix.rs +++ b/tokio/src/signal/unix.rs @@ -407,16 +407,6 @@ impl Signal { } } -cfg_stream! { - impl crate::stream::Stream for Signal { - type Item = (); - - fn poll_next(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - self.poll_recv(cx) - } - } -} - // Work around for abstracting streams internally pub(crate) trait InternalStream: Unpin { fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll>; diff --git a/tokio/src/signal/windows.rs b/tokio/src/signal/windows.rs index 7f2e4862e..43af2906c 100644 --- a/tokio/src/signal/windows.rs +++ b/tokio/src/signal/windows.rs @@ -231,16 +231,6 @@ impl CtrlC { } } -cfg_stream! { - impl crate::stream::Stream for CtrlC { - type Item = (); - - fn poll_next(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - self.poll_recv(cx) - } - } -} - /// Represents a stream which receives "ctrl-break" notifications sent to the process /// via `SetConsoleCtrlHandler`. /// @@ -313,16 +303,6 @@ impl CtrlBreak { } } -cfg_stream! { - impl crate::stream::Stream for CtrlBreak { - type Item = (); - - fn poll_next(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - self.poll_recv(cx) - } - } -} - /// Creates a new stream which receives "ctrl-break" notifications sent to the /// process. /// @@ -351,7 +331,6 @@ pub fn ctrl_break() -> io::Result { mod tests { use super::*; use crate::runtime::Runtime; - use crate::stream::StreamExt; use tokio_test::{assert_ok, assert_pending, assert_ready_ok, task}; @@ -388,7 +367,7 @@ mod tests { super::handler(CTRL_BREAK_EVENT); } - ctrl_break.next().await.unwrap(); + ctrl_break.recv().await.unwrap(); }); } diff --git a/tokio/src/sync/broadcast.rs b/tokio/src/sync/broadcast.rs index ee9aba07a..1b9460031 100644 --- a/tokio/src/sync/broadcast.rs +++ b/tokio/src/sync/broadcast.rs @@ -940,48 +940,6 @@ impl Receiver { let guard = self.recv_ref(None)?; guard.clone_value().ok_or(TryRecvError::Closed) } - - /// Convert the receiver into a `Stream`. - /// - /// The conversion allows using `Receiver` with APIs that require stream - /// values. - /// - /// # Examples - /// - /// ``` - /// use tokio::stream::StreamExt; - /// use tokio::sync::broadcast; - /// - /// #[tokio::main] - /// async fn main() { - /// let (tx, rx) = broadcast::channel(128); - /// - /// tokio::spawn(async move { - /// for i in 0..10_i32 { - /// tx.send(i).unwrap(); - /// } - /// }); - /// - /// // Streams must be pinned to iterate. - /// tokio::pin! { - /// let stream = rx - /// .into_stream() - /// .filter(Result::is_ok) - /// .map(Result::unwrap) - /// .filter(|v| v % 2 == 0) - /// .map(|v| v + 1); - /// } - /// - /// while let Some(i) = stream.next().await { - /// println!("{}", i); - /// } - /// } - /// ``` - #[cfg(feature = "stream")] - #[cfg_attr(docsrs, doc(cfg(feature = "stream")))] - pub fn into_stream(self) -> impl Stream> { - Recv::new(Borrow(self)) - } } impl Drop for Receiver { @@ -1058,31 +1016,6 @@ where } } -cfg_stream! { - use futures_core::Stream; - - impl Stream for Recv - where - R: AsMut>, - T: Clone, - { - type Item = Result; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let (receiver, waiter) = self.project(); - - let guard = match receiver.recv_ref(Some((waiter, cx.waker()))) { - Ok(value) => value, - Err(TryRecvError::Empty) => return Poll::Pending, - Err(TryRecvError::Lagged(n)) => return Poll::Ready(Some(Err(RecvError::Lagged(n)))), - Err(TryRecvError::Closed) => return Poll::Ready(None), - }; - - Poll::Ready(guard.clone_value().map(Ok)) - } - } -} - impl Drop for Recv where R: AsMut>, diff --git a/tokio/src/sync/mpsc/bounded.rs b/tokio/src/sync/mpsc/bounded.rs index c89efdb05..dbe455933 100644 --- a/tokio/src/sync/mpsc/bounded.rs +++ b/tokio/src/sync/mpsc/bounded.rs @@ -11,7 +11,7 @@ cfg_time! { } use std::fmt; -#[cfg(any(feature = "signal", feature = "process", feature = "stream"))] +#[cfg(any(feature = "signal", feature = "process"))] use std::task::{Context, Poll}; /// Send values to the associated `Receiver`. @@ -255,16 +255,6 @@ impl fmt::Debug for Receiver { impl Unpin for Receiver {} -cfg_stream! { - impl crate::stream::Stream for Receiver { - type Item = T; - - fn poll_next(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - self.chan.recv(cx) - } - } -} - impl Sender { pub(crate) fn new(chan: chan::Tx) -> Sender { Sender { chan } diff --git a/tokio/src/sync/mpsc/mod.rs b/tokio/src/sync/mpsc/mod.rs index a2bcf83b0..e7033f6f2 100644 --- a/tokio/src/sync/mpsc/mod.rs +++ b/tokio/src/sync/mpsc/mod.rs @@ -14,10 +14,8 @@ //! Similar to the `mpsc` channels provided by `std`, the channel constructor //! functions provide separate send and receive handles, [`Sender`] and //! [`Receiver`] for the bounded channel, [`UnboundedSender`] and -//! [`UnboundedReceiver`] for the unbounded channel. Both [`Receiver`] and -//! [`UnboundedReceiver`] implement [`Stream`] and allow a task to read -//! values out of the channel. If there is no message to read, the current task -//! will be notified when a new value is sent. [`Sender`] and +//! [`UnboundedReceiver`] for the unbounded channel. If there is no message to read, +//! the current task will be notified when a new value is sent. [`Sender`] and //! [`UnboundedSender`] allow sending values into the channel. If the bounded //! channel is at capacity, the send is rejected and the task will be notified //! when additional capacity is available. In other words, the channel provides @@ -62,7 +60,6 @@ //! //! [`Sender`]: crate::sync::mpsc::Sender //! [`Receiver`]: crate::sync::mpsc::Receiver -//! [`Stream`]: crate::stream::Stream //! [bounded-send]: crate::sync::mpsc::Sender::send() //! [bounded-recv]: crate::sync::mpsc::Receiver::recv() //! [blocking-send]: crate::sync::mpsc::Sender::blocking_send() diff --git a/tokio/src/sync/mpsc/unbounded.rs b/tokio/src/sync/mpsc/unbounded.rs index a88abf956..e8e3faccb 100644 --- a/tokio/src/sync/mpsc/unbounded.rs +++ b/tokio/src/sync/mpsc/unbounded.rs @@ -161,15 +161,6 @@ impl UnboundedReceiver { } } -#[cfg(feature = "stream")] -impl crate::stream::Stream for UnboundedReceiver { - type Item = T; - - fn poll_next(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - self.poll_recv(cx) - } -} - impl UnboundedSender { pub(crate) fn new(chan: chan::Tx) -> UnboundedSender { UnboundedSender { chan } diff --git a/tokio/src/time/interval.rs b/tokio/src/time/interval.rs index c7c58e179..e12a3f339 100644 --- a/tokio/src/time/interval.rs +++ b/tokio/src/time/interval.rs @@ -107,11 +107,6 @@ pub fn interval_at(start: Instant, period: Duration) -> Interval { } /// Stream returned by [`interval`](interval) and [`interval_at`](interval_at). -/// -/// This type only implements the [`Stream`] trait if the "stream" feature is -/// enabled. -/// -/// [`Stream`]: trait@crate::stream::Stream #[derive(Debug)] pub struct Interval { /// Future that completes the next time the `Interval` yields a value. @@ -162,12 +157,3 @@ impl Interval { poll_fn(|cx| self.poll_tick(cx)).await } } - -#[cfg(feature = "stream")] -impl crate::stream::Stream for Interval { - type Item = Instant; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - Poll::Ready(Some(ready!(self.poll_tick(cx)))) - } -} diff --git a/tokio/src/util/mod.rs b/tokio/src/util/mod.rs index 03c2f6bc2..382bbb91a 100644 --- a/tokio/src/util/mod.rs +++ b/tokio/src/util/mod.rs @@ -14,7 +14,7 @@ cfg_io_driver! { ))] pub(crate) mod linked_list; -#[cfg(any(feature = "rt-multi-thread", feature = "macros", feature = "stream"))] +#[cfg(any(feature = "rt-multi-thread", feature = "macros"))] mod rand; cfg_rt! { @@ -32,6 +32,6 @@ cfg_rt_multi_thread! { pub(crate) mod trace; -#[cfg(any(feature = "macros", feature = "stream"))] +#[cfg(any(feature = "macros"))] #[cfg_attr(not(feature = "macros"), allow(unreachable_pub))] pub use rand::thread_rng_n; diff --git a/tokio/src/util/rand.rs b/tokio/src/util/rand.rs index 4b72b4b11..5660103ed 100644 --- a/tokio/src/util/rand.rs +++ b/tokio/src/util/rand.rs @@ -52,7 +52,7 @@ impl FastRand { } // Used by the select macro and `StreamMap` -#[cfg(any(feature = "macros", feature = "stream"))] +#[cfg(any(feature = "macros"))] #[doc(hidden)] #[cfg_attr(not(feature = "macros"), allow(unreachable_pub))] pub fn thread_rng_n(n: u32) -> u32 { diff --git a/tokio/tests/async_send_sync.rs b/tokio/tests/async_send_sync.rs index 2ee385700..c345d2acb 100644 --- a/tokio/tests/async_send_sync.rs +++ b/tokio/tests/async_send_sync.rs @@ -14,8 +14,7 @@ type BoxFutureSync = std::pin::Pin + type BoxFutureSend = std::pin::Pin + Send>>; #[allow(dead_code)] type BoxFuture = std::pin::Pin>>; -#[allow(dead_code)] -type BoxStream = std::pin::Pin>>; + #[allow(dead_code)] type BoxAsyncRead = std::pin::Pin>; #[allow(dead_code)] @@ -222,10 +221,6 @@ async_assert_fn!(tokio::signal::ctrl_c(): Send & Sync); #[cfg(unix)] async_assert_fn!(tokio::signal::unix::Signal::recv(_): Send & Sync); -async_assert_fn!(tokio::stream::empty>(): Send & Sync); -async_assert_fn!(tokio::stream::pending>(): Send & Sync); -async_assert_fn!(tokio::stream::iter(std::vec::IntoIter): Send & Sync); - async_assert_fn!(tokio::sync::Barrier::wait(_): Send & Sync); async_assert_fn!(tokio::sync::Mutex::lock(_): Send & Sync); async_assert_fn!(tokio::sync::Mutex>::lock(_): Send & Sync); @@ -285,13 +280,6 @@ async_assert_fn!(tokio::time::timeout_at(Instant, BoxFutureSend<()>): Send & !Sy async_assert_fn!(tokio::time::timeout_at(Instant, BoxFuture<()>): !Send & !Sync); async_assert_fn!(tokio::time::Interval::tick(_): Send & Sync); -async_assert_fn!(tokio::stream::StreamExt::next(&mut BoxStream<()>): !Unpin); -async_assert_fn!(tokio::stream::StreamExt::try_next(&mut BoxStream>): !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>(&mut BoxStream<()>): !Unpin); - async_assert_fn!(tokio::io::AsyncBufReadExt::read_until(&mut BoxAsyncRead, u8, &mut Vec): !Unpin); async_assert_fn!(tokio::io::AsyncBufReadExt::read_line(&mut BoxAsyncRead, &mut String): !Unpin); async_assert_fn!(tokio::io::AsyncReadExt::read(&mut BoxAsyncRead, &mut [u8]): !Unpin); diff --git a/tokio/tests/fs_dir.rs b/tokio/tests/fs_dir.rs index 6355ef05f..21efe8c0e 100644 --- a/tokio/tests/fs_dir.rs +++ b/tokio/tests/fs_dir.rs @@ -85,35 +85,3 @@ async fn read_inherent() { vec!["aa".to_string(), "bb".to_string(), "cc".to_string()] ); } - -#[tokio::test] -async fn read_stream() { - use tokio::stream::StreamExt; - - let base_dir = tempdir().unwrap(); - - let p = base_dir.path(); - std::fs::create_dir(p.join("aa")).unwrap(); - std::fs::create_dir(p.join("bb")).unwrap(); - std::fs::create_dir(p.join("cc")).unwrap(); - - let files = Arc::new(Mutex::new(Vec::new())); - - let f = files.clone(); - let p = p.to_path_buf(); - - let mut entries = fs::read_dir(p).await.unwrap(); - - while let Some(res) = entries.next().await { - let e = assert_ok!(res); - let s = e.file_name().to_str().unwrap().to_string(); - f.lock().unwrap().push(s); - } - - let mut files = files.lock().unwrap(); - files.sort(); // because the order is not guaranteed - assert_eq!( - *files, - vec!["aa".to_string(), "bb".to_string(), "cc".to_string()] - ); -} diff --git a/tokio/tests/io_lines.rs b/tokio/tests/io_lines.rs index 2f6b3393b..9996d81ca 100644 --- a/tokio/tests/io_lines.rs +++ b/tokio/tests/io_lines.rs @@ -17,19 +17,3 @@ async fn lines_inherent() { assert_eq!(b, ""); assert!(assert_ok!(st.next_line().await).is_none()); } - -#[tokio::test] -async fn lines_stream() { - use tokio::stream::StreamExt; - - let rd: &[u8] = b"hello\r\nworld\n\n"; - let mut st = rd.lines(); - - let b = assert_ok!(st.next().await.unwrap()); - assert_eq!(b, "hello"); - let b = assert_ok!(st.next().await.unwrap()); - assert_eq!(b, "world"); - let b = assert_ok!(st.next().await.unwrap()); - assert_eq!(b, ""); - assert!(st.next().await.is_none()); -} diff --git a/tokio/tests/rt_basic.rs b/tokio/tests/rt_basic.rs index 7b5b622b6..977a838c2 100644 --- a/tokio/tests/rt_basic.rs +++ b/tokio/tests/rt_basic.rs @@ -2,12 +2,16 @@ #![cfg(feature = "full")] use tokio::runtime::Runtime; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::oneshot; use tokio_test::{assert_err, assert_ok}; use std::thread; use std::time::Duration; +mod support { + pub(crate) mod mpsc_stream; +} + #[test] fn spawned_task_does_not_progress_without_block_on() { let (tx, mut rx) = oneshot::channel(); @@ -36,7 +40,7 @@ fn no_extra_poll() { Arc, }; use std::task::{Context, Poll}; - use tokio::stream::{Stream, StreamExt}; + use tokio_stream::{Stream, StreamExt}; pin_project! { struct TrackPolls { @@ -58,8 +62,8 @@ fn no_extra_poll() { } } - let (tx, rx) = mpsc::unbounded_channel(); - let mut rx = TrackPolls { + let (tx, rx) = support::mpsc_stream::unbounded_channel_stream::<()>(); + let rx = TrackPolls { npolls: Arc::new(AtomicUsize::new(0)), s: rx, }; @@ -67,6 +71,9 @@ fn no_extra_poll() { let rt = rt(); + // TODO: could probably avoid this, but why not. + let mut rx = Box::pin(rx); + rt.spawn(async move { while rx.next().await.is_some() {} }); rt.block_on(async { tokio::task::yield_now().await; diff --git a/tokio/tests/support/mpsc_stream.rs b/tokio/tests/support/mpsc_stream.rs new file mode 100644 index 000000000..3df541ff7 --- /dev/null +++ b/tokio/tests/support/mpsc_stream.rs @@ -0,0 +1,29 @@ +#![allow(dead_code)] + +use async_stream::stream; +use tokio::sync::mpsc::{self, Sender, UnboundedSender}; +use tokio_stream::Stream; + +pub fn unbounded_channel_stream() -> (UnboundedSender, impl Stream) { + let (tx, mut rx) = mpsc::unbounded_channel(); + + let stream = stream! { + while let Some(item) = rx.recv().await { + yield item; + } + }; + + (tx, stream) +} + +pub fn channel_stream(size: usize) -> (Sender, impl Stream) { + let (tx, mut rx) = mpsc::channel(size); + + let stream = stream! { + while let Some(item) = rx.recv().await { + yield item; + } + }; + + (tx, stream) +} diff --git a/tokio/tests/sync_broadcast.rs b/tokio/tests/sync_broadcast.rs index 84c77a776..5f79800a7 100644 --- a/tokio/tests/sync_broadcast.rs +++ b/tokio/tests/sync_broadcast.rs @@ -89,46 +89,6 @@ fn send_two_recv() { assert_empty!(rx2); } -#[tokio::test] -async fn send_recv_into_stream_ready() { - use tokio::stream::StreamExt; - - let (tx, rx) = broadcast::channel::(8); - tokio::pin! { - let rx = rx.into_stream(); - } - - assert_ok!(tx.send(1)); - assert_ok!(tx.send(2)); - - assert_eq!(Some(Ok(1)), rx.next().await); - assert_eq!(Some(Ok(2)), rx.next().await); - - drop(tx); - - assert_eq!(None, rx.next().await); -} - -#[tokio::test] -async fn send_recv_into_stream_pending() { - use tokio::stream::StreamExt; - - let (tx, rx) = broadcast::channel::(8); - - tokio::pin! { - let rx = rx.into_stream(); - } - - let mut recv = task::spawn(rx.next()); - assert_pending!(recv.poll()); - - assert_ok!(tx.send(1)); - - assert!(recv.is_woken()); - let val = assert_ready!(recv.poll()); - assert_eq!(val, Some(Ok(1))); -} - #[test] fn send_recv_bounded() { let (tx, mut rx) = broadcast::channel(16); diff --git a/tokio/tests/sync_mpsc.rs b/tokio/tests/sync_mpsc.rs index e64faca2a..b378e6bb9 100644 --- a/tokio/tests/sync_mpsc.rs +++ b/tokio/tests/sync_mpsc.rs @@ -13,6 +13,10 @@ use tokio_test::{ use std::sync::Arc; +mod support { + pub(crate) mod mpsc_stream; +} + trait AssertSend: Send {} impl AssertSend for mpsc::Sender {} impl AssertSend for mpsc::Receiver {} @@ -80,9 +84,10 @@ async fn reserve_disarm() { #[tokio::test] async fn send_recv_stream_with_buffer() { - use tokio::stream::StreamExt; + use tokio_stream::StreamExt; - let (tx, mut rx) = mpsc::channel::(16); + let (tx, rx) = support::mpsc_stream::channel_stream::(16); + let mut rx = Box::pin(rx); tokio::spawn(async move { assert_ok!(tx.send(1).await); @@ -178,9 +183,11 @@ async fn async_send_recv_unbounded() { #[tokio::test] async fn send_recv_stream_unbounded() { - use tokio::stream::StreamExt; + use tokio_stream::StreamExt; - let (tx, mut rx) = mpsc::unbounded_channel::(); + let (tx, rx) = support::mpsc_stream::unbounded_channel_stream::(); + + let mut rx = Box::pin(rx); tokio::spawn(async move { assert_ok!(tx.send(1)); diff --git a/tokio/tests/task_blocking.rs b/tokio/tests/task_blocking.rs index eec19cc16..82bef8a1d 100644 --- a/tokio/tests/task_blocking.rs +++ b/tokio/tests/task_blocking.rs @@ -7,6 +7,10 @@ use tokio_test::assert_ok; use std::thread; use std::time::Duration; +mod support { + pub(crate) mod mpsc_stream; +} + #[tokio::test] async fn basic_blocking() { // Run a few times @@ -165,7 +169,8 @@ fn coop_disabled_in_block_in_place() { .build() .unwrap(); - let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, rx) = support::mpsc_stream::unbounded_channel_stream(); + for i in 0..200 { tx.send(i).unwrap(); } @@ -175,7 +180,7 @@ fn coop_disabled_in_block_in_place() { let jh = tokio::spawn(async move { tokio::task::block_in_place(move || { futures::executor::block_on(async move { - use tokio::stream::StreamExt; + use tokio_stream::StreamExt; assert_eq!(rx.fold(0, |n, _| n + 1).await, 200); }) }) @@ -195,7 +200,8 @@ fn coop_disabled_in_block_in_place_in_block_on() { thread::spawn(move || { let outer = tokio::runtime::Runtime::new().unwrap(); - let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, rx) = support::mpsc_stream::unbounded_channel_stream(); + for i in 0..200 { tx.send(i).unwrap(); } @@ -204,7 +210,7 @@ fn coop_disabled_in_block_in_place_in_block_on() { outer.block_on(async move { tokio::task::block_in_place(move || { futures::executor::block_on(async move { - use tokio::stream::StreamExt; + use tokio_stream::StreamExt; assert_eq!(rx.fold(0, |n, _| n + 1).await, 200); }) }) diff --git a/tokio/tests/tcp_accept.rs b/tokio/tests/tcp_accept.rs index 4c0d6822d..5ffb946f3 100644 --- a/tokio/tests/tcp_accept.rs +++ b/tokio/tests/tcp_accept.rs @@ -46,7 +46,7 @@ use std::sync::{ Arc, }; use std::task::{Context, Poll}; -use tokio::stream::{Stream, StreamExt}; +use tokio_stream::{Stream, StreamExt}; struct TrackPolls<'a> { npolls: Arc, @@ -88,7 +88,7 @@ async fn no_extra_poll() { assert_eq!(npolls.load(SeqCst), 1); let _ = assert_ok!(TcpStream::connect(&addr).await); - accepted_rx.next().await.unwrap(); + accepted_rx.recv().await.unwrap(); // should have been polled twice more: once to yield Some(), then once to yield Pending assert_eq!(npolls.load(SeqCst), 1 + 2); diff --git a/tokio/tests/time_interval.rs b/tokio/tests/time_interval.rs index a07871576..a3c7f0874 100644 --- a/tokio/tests/time_interval.rs +++ b/tokio/tests/time_interval.rs @@ -44,21 +44,6 @@ async fn usage() { assert_pending!(poll_next(&mut i)); } -#[tokio::test] -async fn usage_stream() { - use tokio::stream::StreamExt; - - let start = Instant::now(); - let interval = time::interval(ms(10)); - tokio::pin!(interval); - - for _ in 0..3 { - interval.next().await.unwrap(); - } - - assert!(start.elapsed() > ms(20)); -} - fn poll_next(interval: &mut task::Spawn) -> Poll { interval.enter(|cx, mut interval| { tokio::pin! {