Merge branch 'master' into evanrittenhouse/stable-task-id

README.md

@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
 
 ```toml
 [dependencies]
-tokio = { version = "1.39.3", features = ["full"] }
+tokio = { version = "1.40.0", features = ["full"] }
 ```
 Then, on your main.rs:
 

spellcheck.dic

@@ -1,4 +1,4 @@
-285
+287
 &
 +
 <
@@ -99,6 +99,7 @@ errored
 EWMA
 expirations
 fcntl
+fd
 fd's
 FIFOs
 filename
@@ -151,6 +152,7 @@ metadata
 mio
 Mio
 mio's
+miri
 misconfigured
 mock's
 mpmc

tests-integration/src/bin/test-mem.rs

@@ -1,4 +1,4 @@
-use futures::future::poll_fn;
+use std::future::poll_fn;
 
 fn main() {
     let rt = tokio::runtime::Builder::new_multi_thread()

tests-integration/tests/process_stdio.rs

@@ -10,6 +10,7 @@ use futures::future::{self, FutureExt};
 use std::env;
 use std::io;
 use std::process::{ExitStatus, Stdio};
+use std::task::ready;
 
 fn cat() -> Command {
     let mut cmd = Command::new(env!("CARGO_BIN_EXE_test-cat"));
@@ -205,13 +206,13 @@ async fn vectored_writes() {
         let mut input = Bytes::from_static(b"hello\n").chain(Bytes::from_static(b"world!\n"));
         let mut writes_completed = 0;
 
-        futures::future::poll_fn(|cx| loop {
+        std::future::poll_fn(|cx| loop {
             let mut slices = [IoSlice::new(&[]); 2];
             let vectored = input.chunks_vectored(&mut slices);
             if vectored == 0 {
                 return std::task::Poll::Ready(std::io::Result::Ok(()));
             }
-            let n = futures::ready!(Pin::new(&mut stdin).poll_write_vectored(cx, &slices))?;
+            let n = ready!(Pin::new(&mut stdin).poll_write_vectored(cx, &slices))?;
             writes_completed += 1;
             input.advance(n);
         })

tokio-macros/src/entry.rs

@@ -438,8 +438,9 @@ fn parse_knobs(mut input: ItemFn, is_test: bool, config: FinalConfig) -> TokenSt
     };
 
     let body_ident = quote! { body };
+    // This explicit `return` is intentional. See tokio-rs/tokio#4636
     let last_block = quote_spanned! {last_stmt_end_span=>
-        #[allow(clippy::expect_used, clippy::diverging_sub_expression)]
+        #[allow(clippy::expect_used, clippy::diverging_sub_expression, clippy::needless_return)]
         {
             return #rt
                 .enable_all()

tokio-stream/CHANGELOG.md

@@ -1,3 +1,15 @@
+# 0.1.16 (September 5th, 2024)
+
+This release bumps the MSRV of tokio-stream to 1.70.
+
+- stream: add `next_many` and `poll_next_many` to `StreamMap` ([#6409])
+- stream: make stream adapters public ([#6658])
+- readme: add readme for tokio-stream ([#6456])
+
+[#6409]: https://github.com/tokio-rs/tokio/pull/6409
+[#6658]: https://github.com/tokio-rs/tokio/pull/6658
+[#6456]: https://github.com/tokio-rs/tokio/pull/6456
+
 # 0.1.15 (March 14th, 2024)
 
 This release bumps the MSRV of tokio-stream to 1.63.

tokio-stream/Cargo.toml

@@ -4,7 +4,7 @@ name = "tokio-stream"
 # - Remove path dependencies
 # - Update CHANGELOG.md.
 # - Create "tokio-stream-0.1.x" git tag.
-version = "0.1.15"
+version = "0.1.16"
 edition = "2021"
 rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]

tokio-stream/src/lib.rs

@@ -74,9 +74,6 @@
 #[macro_use]
 mod macros;
 
-mod poll_fn;
-pub(crate) use poll_fn::poll_fn;
-
 pub mod wrappers;
 
 mod stream_ext;

tokio-stream/src/macros.rs

@@ -57,12 +57,3 @@ macro_rules! cfg_signal {
         )*
     }
 }
-
-macro_rules! ready {
-    ($e:expr $(,)?) => {
-        match $e {
-            std::task::Poll::Ready(t) => t,
-            std::task::Poll::Pending => return std::task::Poll::Pending,
-        }
-    };
-}

tokio-stream/src/stream_ext/all.rs

@@ -3,7 +3,7 @@ use crate::Stream;
 use core::future::Future;
 use core::marker::PhantomPinned;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {
@@ -42,7 +42,7 @@ where
 
         // Take a maximum of 32 items from the stream before yielding.
         for _ in 0..32 {
-            match futures_core::ready!(stream.as_mut().poll_next(cx)) {
+            match ready!(stream.as_mut().poll_next(cx)) {
                 Some(v) => {
                     if !(me.f)(v) {
                         return Poll::Ready(false);

tokio-stream/src/stream_ext/any.rs

@@ -3,7 +3,7 @@ use crate::Stream;
 use core::future::Future;
 use core::marker::PhantomPinned;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {
@@ -42,7 +42,7 @@ where
 
         // Take a maximum of 32 items from the stream before yielding.
         for _ in 0..32 {
-            match futures_core::ready!(stream.as_mut().poll_next(cx)) {
+            match ready!(stream.as_mut().poll_next(cx)) {
                 Some(v) => {
                     if (me.f)(v) {
                         return Poll::Ready(true);

tokio-stream/src/stream_ext/chain.rs

@@ -2,7 +2,7 @@ use crate::stream_ext::Fuse;
 use crate::Stream;
 
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_ext/chunks_timeout.rs

@@ -4,7 +4,7 @@ use tokio::time::{sleep, Sleep};
 
 use core::future::Future;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 use std::time::Duration;
 

tokio-stream/src/stream_ext/collect.rs

@@ -4,7 +4,7 @@ use core::future::Future;
 use core::marker::PhantomPinned;
 use core::mem;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 // Do not export this struct until `FromStream` can be unsealed.

tokio-stream/src/stream_ext/filter.rs

@@ -2,7 +2,7 @@ use crate::Stream;
 
 use core::fmt;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_ext/filter_map.rs

@@ -2,7 +2,7 @@ use crate::Stream;
 
 use core::fmt;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_ext/fold.rs

@@ -3,7 +3,7 @@ use crate::Stream;
 use core::future::Future;
 use core::marker::PhantomPinned;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_ext/fuse.rs

@@ -2,7 +2,7 @@ use crate::Stream;
 
 use pin_project_lite::pin_project;
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 
 pin_project! {
     /// Stream returned by [`fuse()`][super::StreamExt::fuse].

tokio-stream/src/stream_ext/peekable.rs

@@ -8,7 +8,7 @@ use crate::stream_ext::Fuse;
 use crate::StreamExt;
 
 pin_project! {
-    /// Stream returned by the [`chain`](super::StreamExt::peekable) method.
+    /// Stream returned by the [`peekable`](super::StreamExt::peekable) method.
     pub struct Peekable<T: Stream> {
         peek: Option<T::Item>,
         #[pin]

tokio-stream/src/stream_ext/skip.rs

@@ -2,7 +2,7 @@ use crate::Stream;
 
 use core::fmt;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_ext/skip_while.rs

@@ -2,7 +2,7 @@ use crate::Stream;
 
 use core::fmt;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_ext/throttle.rs

@@ -5,7 +5,7 @@ use tokio::time::{Duration, Instant, Sleep};
 
 use std::future::Future;
 use std::pin::Pin;
-use std::task::{self, Poll};
+use std::task::{self, ready, Poll};
 
 use pin_project_lite::pin_project;
 

tokio-stream/src/stream_ext/timeout.rs

@@ -4,7 +4,7 @@ use tokio::time::{Instant, Sleep};
 
 use core::future::Future;
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 use std::fmt;
 use std::time::Duration;

tokio-stream/src/stream_ext/timeout_repeating.rs

@@ -3,7 +3,7 @@ use crate::{Elapsed, Stream};
 use tokio::time::Interval;
 
 use core::pin::Pin;
-use core::task::{Context, Poll};
+use core::task::{ready, Context, Poll};
 use pin_project_lite::pin_project;
 
 pin_project! {

tokio-stream/src/stream_map.rs

@@ -1,9 +1,10 @@
-use crate::{poll_fn, Stream};
+use crate::Stream;
 
 use std::borrow::Borrow;
+use std::future::poll_fn;
 use std::hash::Hash;
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 
 /// Combine many streams into one, indexing each source stream with a unique
 /// key.

tokio-stream/src/wrappers/broadcast.rs

@@ -6,12 +6,12 @@ use futures_core::Stream;
 use tokio_util::sync::ReusableBoxFuture;
 
 use std::fmt;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 
 /// A wrapper around [`tokio::sync::broadcast::Receiver`] that implements [`Stream`].
 ///
 /// [`tokio::sync::broadcast::Receiver`]: struct@tokio::sync::broadcast::Receiver
-/// [`Stream`]: trait@crate::Stream
+/// [`Stream`]: trait@futures_core::Stream
 #[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
 pub struct BroadcastStream<T> {
     inner: ReusableBoxFuture<'static, (Result<T, RecvError>, Receiver<T>)>,

tokio-stream/src/wrappers/watch.rs

@@ -5,7 +5,7 @@ use futures_core::Stream;
 use tokio_util::sync::ReusableBoxFuture;
 
 use std::fmt;
-use std::task::{Context, Poll};
+use std::task::{ready, Context, Poll};
 use tokio::sync::watch::error::RecvError;
 
 /// A wrapper around [`tokio::sync::watch::Receiver`] that implements [`Stream`].

tokio-test/src/io.rs

@@ -23,13 +23,13 @@ use tokio::sync::mpsc;
 use tokio::time::{self, Duration, Instant, Sleep};
 use tokio_stream::wrappers::UnboundedReceiverStream;
 
-use futures_core::{ready, Stream};
+use futures_core::Stream;
 use std::collections::VecDeque;
 use std::fmt;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
-use std::task::{self, Poll, Waker};
+use std::task::{self, ready, Poll, Waker};
 use std::{cmp, io};
 
 /// An I/O object that follows a predefined script.

tokio-test/src/stream_mock.rs

@@ -36,10 +36,10 @@
 
 use std::collections::VecDeque;
 use std::pin::Pin;
-use std::task::Poll;
+use std::task::{ready, Poll};
 use std::time::Duration;
 
-use futures_core::{ready, Stream};
+use futures_core::Stream;
 use std::future::Future;
 use tokio::time::{sleep_until, Instant, Sleep};