tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-28 00:00:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

ci: unfreeze wasm tests from rustc 1.88.0 (#7537)

Browse Source
This commit is contained in:
Lucas Black
2025-09-26 21:29:01 +08:00
committed by GitHub
parent bce76c515f
commit 8ccf2fb92e
74 changed files with 3496 additions and 3180 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio-stream/src/empty.rs
+5 -5
View File
@@ -26,12 +26,12 @@ unsafe impl<T> Sync for Empty<T> {}
/// ```
/// use tokio_stream::{self as stream, StreamExt};
///
/// #[tokio::main]
/// async fn main() {
/// let mut none = stream::empty::<i32>();
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// let mut none = stream::empty::<i32>();
///
/// assert_eq!(None, none.next().await);
/// }
/// assert_eq!(None, none.next().await);
/// # }
/// ```
pub const fn empty<T>() -> Empty<T> {
Empty(PhantomData)
tokio-stream/src/lib.rs
+6 -6
View File
@@ -34,14 +34,14 @@
//! ```rust
//! use tokio_stream::{self as stream, StreamExt};
//!
//! #[tokio::main]
//! async fn main() {
//! let mut stream = stream::iter(vec![0, 1, 2]);
//! # #[tokio::main(flavor = "current_thread")]
//! # async fn main() {
//! let mut stream = stream::iter(vec![0, 1, 2]);
//!
//! while let Some(value) = stream.next().await {
//! println!("Got {}", value);
//! }
//! while let Some(value) = stream.next().await {
//! println!("Got {}", value);
//! }
//! # }
//! ```
//!
//! # Returning a Stream from a function
tokio-stream/src/once.rs
+8 -8
View File
@@ -22,16 +22,16 @@ impl<I> Unpin for Once<I> {}
/// ```
/// use tokio_stream::{self as stream, StreamExt};
///
/// #[tokio::main]
/// async fn main() {
/// // one is the loneliest number
/// let mut one = stream::once(1);
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// // one is the loneliest number
/// let mut one = stream::once(1);
///
/// assert_eq!(Some(1), one.next().await);
/// assert_eq!(Some(1), one.next().await);
///
/// // just one, that's all we get
/// assert_eq!(None, one.next().await);
/// }
/// // just one, that's all we get
/// assert_eq!(None, one.next().await);
/// # }
/// ```
pub fn once<T>(value: T) -> Once<T> {
Once {
tokio-stream/src/stream_close.rs
+12 -12
View File
@@ -17,20 +17,20 @@ pin_project! {
/// ```
/// use tokio_stream::{StreamExt, StreamMap, StreamNotifyClose};
///
/// #[tokio::main]
/// async fn main() {
/// let mut map = StreamMap::new();
/// let stream = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
/// let stream2 = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
/// map.insert(0, stream);
/// map.insert(1, stream2);
/// while let Some((key, val)) = map.next().await {
/// match val {
/// Some(val) => println!("got {val:?} from stream {key:?}"),
/// None => println!("stream {key:?} closed"),
/// }
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// let mut map = StreamMap::new();
/// let stream = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
/// let stream2 = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
/// map.insert(0, stream);
/// map.insert(1, stream2);
/// while let Some((key, val)) = map.next().await {
/// match val {
/// Some(val) => println!("got {val:?} from stream {key:?}"),
/// None => println!("stream {key:?} closed"),
/// }
/// }
/// # }
/// ```
#[must_use = "streams do nothing unless polled"]
pub struct StreamNotifyClose<S> {
tokio-stream/src/stream_ext.rs
+60 -56
View File
@@ -129,7 +129,7 @@ pub trait StreamExt: Stream {
/// # Examples
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
///
@@ -171,8 +171,9 @@ pub trait StreamExt: Stream {
/// # Examples
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
///
/// use tokio_stream::{self as stream, StreamExt};
///
/// let mut stream = stream::iter(vec![Ok(1), Ok(2), Err("nope")]);
@@ -203,7 +204,7 @@ pub trait StreamExt: Stream {
/// # Examples
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
///
@@ -239,7 +240,7 @@ pub trait StreamExt: Stream {
/// # Examples
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
///
@@ -283,7 +284,7 @@ pub trait StreamExt: Stream {
/// # Examples
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
///
@@ -418,7 +419,7 @@ pub trait StreamExt: Stream {
/// # Examples
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
///
@@ -454,7 +455,7 @@ pub trait StreamExt: Stream {
///
/// # Examples
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
///
@@ -514,7 +515,10 @@ pub trait StreamExt: Stream {
/// }
/// }
///
/// # /*
/// #[tokio::main]
/// # */
/// # #[tokio::main(flavor = "current_thread")]
/// async fn main() {
/// let mut stream = Alternate { state: 0 };
///
@@ -551,7 +555,7 @@ pub trait StreamExt: Stream {
/// # Examples
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
///
@@ -580,7 +584,7 @@ pub trait StreamExt: Stream {
/// # Examples
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
///
@@ -606,7 +610,7 @@ pub trait StreamExt: Stream {
/// # Examples
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
///
@@ -637,7 +641,7 @@ pub trait StreamExt: Stream {
/// # Examples
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
/// let mut stream = stream::iter(vec![1,2,3,4,1]).skip_while(|x| *x < 3);
@@ -680,7 +684,7 @@ pub trait StreamExt: Stream {
/// Basic usage:
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
///
@@ -695,7 +699,7 @@ pub trait StreamExt: Stream {
/// Stopping at the first `false`:
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
///
@@ -739,7 +743,7 @@ pub trait StreamExt: Stream {
/// Basic usage:
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
///
@@ -754,7 +758,7 @@ pub trait StreamExt: Stream {
/// Stopping at the first `true`:
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
///
@@ -787,21 +791,21 @@ pub trait StreamExt: Stream {
/// ```
/// use tokio_stream::{self as stream, StreamExt};
///
/// #[tokio::main]
/// async fn main() {
/// let one = stream::iter(vec![1, 2, 3]);
/// let two = stream::iter(vec![4, 5, 6]);
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// let one = stream::iter(vec![1, 2, 3]);
/// let two = stream::iter(vec![4, 5, 6]);
///
/// let mut stream = one.chain(two);
/// let mut stream = one.chain(two);
///
/// assert_eq!(stream.next().await, Some(1));
/// assert_eq!(stream.next().await, Some(2));
/// assert_eq!(stream.next().await, Some(3));
/// assert_eq!(stream.next().await, Some(4));
/// assert_eq!(stream.next().await, Some(5));
/// assert_eq!(stream.next().await, Some(6));
/// assert_eq!(stream.next().await, None);
/// }
/// assert_eq!(stream.next().await, Some(1));
/// assert_eq!(stream.next().await, Some(2));
/// assert_eq!(stream.next().await, Some(3));
/// assert_eq!(stream.next().await, Some(4));
/// assert_eq!(stream.next().await, Some(5));
/// assert_eq!(stream.next().await, Some(6));
/// assert_eq!(stream.next().await, None);
/// # }
/// ```
fn chain<U>(self, other: U) -> Chain<Self, U>
where
@@ -823,7 +827,7 @@ pub trait StreamExt: Stream {
/// # Examples
/// Basic usage:
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, *};
///
@@ -874,16 +878,16 @@ pub trait StreamExt: Stream {
/// ```
/// use tokio_stream::{self as stream, StreamExt};
///
/// #[tokio::main]
/// async fn main() {
/// let doubled: Vec<i32> =
/// stream::iter(vec![1, 2, 3])
/// .map(|x| x * 2)
/// .collect()
/// .await;
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// let doubled: Vec<i32> =
/// stream::iter(vec![1, 2, 3])
/// .map(|x| x * 2)
/// .collect()
/// .await;
///
/// assert_eq!(vec![2, 4, 6], doubled);
/// }
/// assert_eq!(vec![2, 4, 6], doubled);
/// # }
/// ```
///
/// Collecting a stream of `Result` values
@@ -891,26 +895,26 @@ pub trait StreamExt: Stream {
/// ```
/// use tokio_stream::{self as stream, StreamExt};
///
/// #[tokio::main]
/// async fn main() {
/// // A stream containing only `Ok` values will be collected
/// let values: Result<Vec<i32>, &str> =
/// stream::iter(vec![Ok(1), Ok(2), Ok(3)])
/// .collect()
/// .await;
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// // A stream containing only `Ok` values will be collected
/// let values: Result<Vec<i32>, &str> =
/// stream::iter(vec![Ok(1), Ok(2), Ok(3)])
/// .collect()
/// .await;
///
/// assert_eq!(Ok(vec![1, 2, 3]), values);
/// assert_eq!(Ok(vec![1, 2, 3]), values);
///
/// // A stream containing `Err` values will return the first error.
/// let results = vec![Ok(1), Err("no"), Ok(2), Ok(3), Err("nein")];
/// // A stream containing `Err` values will return the first error.
/// let results = vec![Ok(1), Err("no"), Ok(2), Ok(3), Err("nein")];
///
/// let values: Result<Vec<i32>, &str> =
/// stream::iter(results)
/// .collect()
/// .await;
/// let values: Result<Vec<i32>, &str> =
/// stream::iter(results)
/// .collect()
/// .await;
///
/// assert_eq!(Err("no"), values);
/// }
/// assert_eq!(Err("no"), values);
/// # }
/// ```
fn collect<T>(self) -> Collect<Self, T>
where
@@ -945,7 +949,7 @@ pub trait StreamExt: Stream {
/// Suppose we have a stream `int_stream` that yields 3 numbers (1, 2, 3):
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
/// use std::time::Duration;
@@ -1031,7 +1035,7 @@ pub trait StreamExt: Stream {
/// Suppose we have a stream `int_stream` that yields 3 numbers (1, 2, 3):
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{self as stream, StreamExt};
/// use std::time::Duration;
tokio-stream/src/stream_map.rs
+60 -60
View File
@@ -65,57 +65,57 @@ use std::task::{ready, Context, Poll};
/// use tokio::sync::mpsc;
/// use std::pin::Pin;
///
/// #[tokio::main]
/// async fn main() {
/// let (tx1, mut rx1) = mpsc::channel::<usize>(10);
/// let (tx2, mut rx2) = mpsc::channel::<usize>(10);
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// let (tx1, mut rx1) = mpsc::channel::<usize>(10);
/// let (tx2, mut rx2) = mpsc::channel::<usize>(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<Box<dyn Stream<Item = usize> + Send>>;
///
/// let rx2 = Box::pin(async_stream::stream! {
/// while let Some(item) = rx2.recv().await {
/// yield item;
/// }
/// }) as Pin<Box<dyn Stream<Item = usize> + Send>>;
///
/// tokio::spawn(async move {
/// tx1.send(1).await.unwrap();
///
/// // This value will never be received. The send may or may not return
/// // `Err` depending on if the remote end closed first or not.
/// let _ = tx1.send(2).await;
/// });
///
/// tokio::spawn(async move {
/// tx2.send(3).await.unwrap();
/// let _ = tx2.send(4).await;
/// });
///
/// let mut map = StreamMap::new();
///
/// // Insert both streams
/// map.insert("one", rx1);
/// map.insert("two", rx2);
///
/// // Read twice
/// for _ in 0..2 {
/// let (key, val) = map.next().await.unwrap();
///
/// if key == "one" {
/// assert_eq!(val, 1);
/// } else {
/// assert_eq!(val, 3);
/// }
///
/// // Remove the stream to prevent reading the next value
/// map.remove(key);
/// // Convert the channels to a `Stream`.
/// let rx1 = Box::pin(async_stream::stream! {
/// while let Some(item) = rx1.recv().await {
/// yield item;
/// }
/// }) as Pin<Box<dyn Stream<Item = usize> + Send>>;
///
/// let rx2 = Box::pin(async_stream::stream! {
/// while let Some(item) = rx2.recv().await {
/// yield item;
/// }
/// }) as Pin<Box<dyn Stream<Item = usize> + Send>>;
///
/// tokio::spawn(async move {
/// tx1.send(1).await.unwrap();
///
/// // This value will never be received. The send may or may not return
/// // `Err` depending on if the remote end closed first or not.
/// let _ = tx1.send(2).await;
/// });
///
/// tokio::spawn(async move {
/// tx2.send(3).await.unwrap();
/// let _ = tx2.send(4).await;
/// });
///
/// let mut map = StreamMap::new();
///
/// // Insert both streams
/// map.insert("one", rx1);
/// map.insert("two", rx2);
///
/// // Read twice
/// for _ in 0..2 {
/// let (key, val) = map.next().await.unwrap();
///
/// if key == "one" {
/// assert_eq!(val, 1);
/// } else {
/// assert_eq!(val, 3);
/// }
///
/// // Remove the stream to prevent reading the next value
/// map.remove(key);
/// }
/// # }
/// ```
///
/// This example models a read-only client to a chat system with channels. The
@@ -185,20 +185,20 @@ use std::task::{ready, Context, Poll};
/// ```
/// use tokio_stream::{StreamExt, StreamMap, StreamNotifyClose};
///
/// #[tokio::main]
/// async fn main() {
/// let mut map = StreamMap::new();
/// let stream = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
/// let stream2 = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
/// map.insert(0, stream);
/// map.insert(1, stream2);
/// while let Some((key, val)) = map.next().await {
/// match val {
/// Some(val) => println!("got {val:?} from stream {key:?}"),
/// None => println!("stream {key:?} closed"),
/// }
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// let mut map = StreamMap::new();
/// let stream = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
/// let stream2 = StreamNotifyClose::new(tokio_stream::iter(vec![0, 1]));
/// map.insert(0, stream);
/// map.insert(1, stream2);
/// while let Some((key, val)) = map.next().await {
/// match val {
/// Some(val) => println!("got {val:?} from stream {key:?}"),
/// None => println!("stream {key:?} closed"),
/// }
/// }
/// # }
/// ```
#[derive(Debug)]
tokio-stream/src/wrappers/tcp_listener.rs
+3
View File
@@ -11,6 +11,8 @@ use tokio::net::{TcpListener, TcpStream};
/// Accept connections from both IPv4 and IPv6 listeners in the same loop:
///
/// ```no_run
/// # #[cfg(not(target_family = "wasm"))]
/// # {
/// use std::net::{Ipv4Addr, Ipv6Addr};
///
/// use tokio::net::TcpListener;
@@ -31,6 +33,7 @@ use tokio::net::{TcpListener, TcpStream};
/// }
/// # Ok(())
/// # }
/// # }
/// ```
///
/// [`TcpListener`]: struct@tokio::net::TcpListener
tokio-stream/src/wrappers/watch.rs
+3 -3
View File
@@ -17,7 +17,7 @@ use tokio::sync::watch::error::RecvError;
/// # Examples
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{StreamExt, wrappers::WatchStream};
/// use tokio::sync::watch;
@@ -33,7 +33,7 @@ use tokio::sync::watch::error::RecvError;
/// ```
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use tokio_stream::{StreamExt, wrappers::WatchStream};
/// use tokio::sync::watch;
@@ -51,7 +51,7 @@ use tokio::sync::watch::error::RecvError;
/// Example with [`WatchStream<T>::from_changes`]:
///
/// ```
/// # #[tokio::main]
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// use futures::future::FutureExt;
/// use tokio::sync::watch;