Tokio Util

https://docs.rs/tokio-util/latest
Admin
Tokio Util is a Rust crate providing utilities for working with Tokio, including codecs,...
Tokens:776,832
Snippets:5,495
Trust Score:9.7
Update:1 week ago
Show doc for...
Tokens:
Raw
### Example: Turn an AsyncRead into a stream of BytesMut using BytesCodec

Source: https://docs.rs/tokio-util/latest/src/tokio_util/codec/bytes_codec.rs.html?search=

Shows how to use FramedRead with BytesCodec to read bytes from a file. The example is a doctest with hidden setup code that opens a file and reads it asynchronously.

```rust
/// ```
/// # mod hidden {
/// # #[allow(unused_imports)]
/// use tokio::fs::File;
/// # }
/// use tokio::io::AsyncRead;
/// use tokio_util::codec::{FramedRead, BytesCodec};
///
/// # enum File {}
/// # impl File {
/// #     async fn open(_name: &str) -> Result<impl AsyncRead, std::io::Error> {
/// #         use std::io::Cursor;
/// #         Ok(Cursor::new(vec![0, 1, 2, 3, 4, 5]))
/// #     }
/// # }
/// #
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() -> Result<(), std::io::Error> {
/// let my_async_read = File::open("filename.txt").await?;
/// let my_stream_of_bytes = FramedRead::new(my_async_read, BytesCodec::new());
/// # Ok(())
/// # }
/// ```
```

--------------------------------

### DelayQueue::reserve example

Source: https://docs.rs/tokio-util/latest/src/tokio_util/time/delay_queue.rs.html

Shows how to reserve additional capacity in a DelayQueue. The example inserts an item and then reserves 10 slots, asserting that capacity is at least 11.

```rust
use tokio_util::time::DelayQueue;
use std::time::Duration;

# #[tokio::main(flavor = "current_thread")]
# async fn main() {
let mut delay_queue = DelayQueue::new();

delay_queue.insert("hello", Duration::from_secs(10));
delay_queue.reserve(10);

assert!(delay_queue.capacity() >= 11);
# }
```

--------------------------------

### spawn_pinned method example with Rc and assert_eq

Source: https://docs.rs/tokio-util/latest/src/tokio_util/task/spawn_pinned.rs.html?search=u32+-%3E+bool

Example from the spawn_pinned method documentation showing how to create a pool of 1 thread and spawn a !Send future that holds an Rc. The output is asserted to be "test". This example is wrapped in a cfg(not(target_family = "wasm")) block.

```rust
# #[cfg(not(target_family = "wasm"))]
# {
use std::rc::Rc;
use tokio_util::task::LocalPoolHandle;

#[tokio::main]
async fn main() {
    // Create the local pool
    let pool = LocalPoolHandle::new(1);

    // Spawn a !Send future onto the pool and await it
    let output = pool
        .spawn_pinned(|| {
            // Rc is !Send + !Sync
            let local_data = Rc::new("test");

            // This future holds an Rc, so it is !Send
            async move { local_data.to_string() }
        })
        .await
        .unwrap();

    assert_eq!(output, "test");
}
# }
```

--------------------------------

### JoinMap::try_join_next example

Source: https://docs.rs/tokio-util/latest/src/tokio_util/task/join_map.rs.html?search=

Example showing how to use try_join_next to poll for a completed task, returning the key and result.

```rust
use tokio_util::task::JoinMap;

# #[tokio::main(flavor = "current_thread")]
# async fn main() {
let mut map = JoinMap::new();
map.spawn("answer", async { 42 });

let (key, res) = loop {
    if let Some(joined) = map.try_join_next() {
        break joined;
    }
```

--------------------------------

### LengthDelimitedCodec usage example

Source: https://docs.rs/tokio-util/latest/src/tokio_util/codec/length_delimited.rs.html

Shows how to create a LengthDelimitedCodec, encode a frame with a length header, and decode it back. The example uses a simple in-memory buffer to demonstrate the encoding and decoding process.

```rust
use tokio_util::codec::LengthDelimitedCodec;

let codec = LengthDelimitedCodec::new();

let mut buf = BytesMut::new();
codec.encode(Bytes::from("hello world"), &mut buf).unwrap();

let mut buf = &mut buf[..];
let frame = codec.decode(&mut buf).unwrap().unwrap();
assert_eq!(frame, Bytes::from("hello world"));
```

--------------------------------

### Get the capacity of a DelayQueue with capacity()

Source: https://docs.rs/tokio-util/latest/src/tokio_util/time/delay_queue.rs.html

Shows how to use capacity() to get the number of elements the queue can hold without reallocating. The example creates a queue with capacity 10 and asserts the capacity is 10.

```rust
use tokio_util::time::DelayQueue;

let delay_queue: DelayQueue<i32> = DelayQueue::with_capacity(10);
assert_eq!(delay_queue.capacity(), 10);
```

--------------------------------

### Example 1: Adapt `tokio::net::TcpStream` to `futures::io::AsyncReadExt`

Source: https://docs.rs/tokio-util/latest/src/tokio_util/compat.rs.html?search=

Shows how to use `compat()` to adapt a `tokio::net::TcpStream` so it can be read with `futures::io::AsyncReadExt::read`. The example spawns a client that writes "Hello World" and then reads it on the server side. Requires the `compat` feature and is not run on wasm targets.

```rust
# #[cfg(not(target_family = "wasm"))]
# {
use tokio::net::{TcpListener, TcpStream};
use tokio::io::AsyncWriteExt;
use tokio_util::compat::TokioAsyncReadCompatExt;
use futures::io::AsyncReadExt;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:8081").await?;

    tokio::spawn(async {
        let mut client = TcpStream::connect("127.0.0.1:8081").await.unwrap();
        client.write_all(b"Hello World").await.unwrap();
    });

    let (stream, _) = listener.accept().await?;

    // Adapt `tokio::TcpStream` to be used with `futures::io::AsyncReadExt`
    let mut compat_stream = stream.compat();
    let mut buffer = [0; 20];
    let n = compat_stream.read(&mut buffer).await?;
    println!("Received: {}", String::from_utf8_lossy(&buffer[..n]));

    Ok(())
}
# }
```

--------------------------------

### take

Source: https://docs.rs/tokio-util/latest/tokio_util/io/simplex/struct.Receiver.html?search=u32+-%3E+bool

Creates an adaptor which reads at most `limit` bytes from the source.

```APIDOC
## take

### Description
Creates an adaptor which reads at most `limit` bytes from it.

### Method
take

### Signature
fn take(self, limit: u64) -> Take<Self>
where Self: Sized

### Parameters
- **limit** (u64) - Required - The maximum number of bytes to read.

### Returns
- **Take<Self>** - An adaptor that limits the number of bytes read.

### Notes
This method is part of the `AsyncReadExt` trait.
```

--------------------------------

### Get maximum chunk length

Source: https://docs.rs/tokio-util/latest/src/tokio_util/codec/any_delimiter_codec.rs.html?search=std%3A%3Avec

Example showing how to retrieve the maximum chunk length from an AnyDelimiterCodec instance.

```rust
use tokio_util::codec::AnyDelimiterCodec;

let codec = AnyDelimiterCodec::new(b",;\n".to_vec(), b";".to_vec());
```

--------------------------------

### fn take(self, limit: u64) -> Take<Self>

Source: https://docs.rs/tokio-util/latest/tokio_util/io/struct.StreamReader.html?search=u32+-%3E+bool

Creates an adaptor which reads at most `limit` bytes from it.

```APIDOC
## fn take(self, limit: u64) -> Take<Self>

### Description
Creates an adaptor which reads at most `limit` bytes from it.

### Method
take

### Signature
fn take(self, limit: u64) -> Take<Self>
where Self: Sized

### Parameters
- **self** (Self) - The reader to limit.
- **limit** (u64) - The maximum number of bytes to read.

### Returns
- **Take<Self>** - An adaptor that reads at most `limit` bytes.
```

--------------------------------

### Spawn multiple tasks and wait for them using JoinMap

Source: https://docs.rs/tokio-util/latest/src/tokio_util/task/join_map.rs.html?search=

Shows how to create a JoinMap, spawn tasks with keys, and use join_next() to await completion. The example uses a current_thread Tokio runtime and asserts that all tasks complete successfully.

```rust
use tokio_util::task::JoinMap;

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let mut map = JoinMap::new();

    for i in 0..10 {
        // Spawn a task on the `JoinMap` with `i` as its key.
        map.spawn(i, async move { /* ... */ });
    }

    let mut seen = [false; 10];

    // When a task completes, `join_next` returns the task's key along
    // with its output.
    while let Some((key, res)) = map.join_next().await {
        seen[key] = true;
        assert!(res.is_ok(), "task {} completed successfully!", key);
    }

    for i in 0..10 {
        assert!(seen[i]);
    }
}
```

--------------------------------

### Get the number of items in a DelayQueue with len()

Source: https://docs.rs/tokio-util/latest/src/tokio_util/time/delay_queue.rs.html

Demonstrates len() to get the number of elements currently in the queue. The example creates a queue with capacity 10, checks len is 0, inserts an item, and checks len is 1.

```rust
use tokio_util::time::DelayQueue;
use std::time::Duration;

# #[tokio::main(flavor = "current_thread")]
# async fn main() {
let mut delay_queue: DelayQueue<i32> = DelayQueue::with_capacity(10);
assert_eq!(delay_queue.len(), 0);
delay_queue.insert(3, Duration::from_secs(5));
assert_eq!(delay_queue.len(), 1);
# }
```

--------------------------------

### Using TaskTracker for graceful shutdown with CancellationToken

Source: https://docs.rs/tokio-util/latest/src/tokio_util/task/task_tracker.rs.html

Shows how to create a TaskTracker, spawn tracked tasks, close the tracker, cancel tasks via a CancellationToken, and wait for all tasks to exit. The example includes cleanup logic and a background task that triggers shutdown after a delay.

```rust
use std::time::Duration;
use tokio::time;
use tokio_util::sync::CancellationToken;
use tokio_util::task::TaskTracker;

#[tokio::main]
async fn main() {
    let tracker = TaskTracker::new();
    let token = CancellationToken::new();

    for i in 0..10 {
        let token = token.clone();
        tracker.spawn(async move {
            // ...
            tokio::select! {
                _ = token.cancelled() => {
                    // The token was cancelled, we can now shut down
                }
                _ = async {
                    // ...
                } => {
                    // ...
                }
            }
        });
    }

    // ...
}
```

--------------------------------

### Get the deadline of an item with deadline()

Source: https://docs.rs/tokio-util/latest/tokio_util/time/delay_queue/struct.DelayQueue.html

Returns the deadline (Instant) for a given key. The example inserts two items and asserts that the first deadline is earlier than the second.

```rust
use tokio_util::time::DelayQueue;
use std::time::Duration;

let mut delay_queue = DelayQueue::new();

let key1 = delay_queue.insert("foo", Duration::from_secs(5));
let key2 = delay_queue.insert("bar", Duration::from_secs(10));

assert!(delay_queue.deadline(&key1) < delay_queue.deadline(&key2));
```

--------------------------------

### Spawn tasks and wait for them to exit with TaskTracker

Source: https://docs.rs/tokio-util/latest/src/tokio_util/task/task_tracker.rs.html

Simple example showing how to create a TaskTracker, spawn tasks, close it, and wait for all tasks to finish. This is a basic use case where JoinSet might be preferred.

```rust
use tokio_util::task::TaskTracker;

# #[tokio::main(flavor = "current_thread")]
# async fn main() {
let tracker = TaskTracker::new();

for i in 0..10 {
    tracker.spawn(async move {
        println!("Task {} is running!", i);
    });
}
// Once we spawned everything, we close the tracker.
tracker.close();

// Wait for everything to finish.
tracker.wait().await;

println!("This is printed after all of the tasks.");
# }
```

--------------------------------

### Example: Turn an AsyncRead into a stream of BytesMut using FramedRead and BytesCodec

Source: https://docs.rs/tokio-util/latest/src/tokio_util/codec/bytes_codec.rs.html?search=std%3A%3Avec

Shows how to use BytesCodec with FramedRead to convert an AsyncRead into a stream of Result<BytesMut, io::Error>. The example opens a file asynchronously and creates a framed stream. The hidden lines are part of the doctest setup and are not shown in the rendered documentation.

```rust
/// # Example
///
/// Turn an [`AsyncRead`] into a stream of `Result<`[`BytesMut`]`, `[`Error`]`>`.
///
/// [`AsyncRead`]: tokio::io::AsyncRead
/// [`BytesMut`]: bytes::BytesMut
/// [`Error`]: std::io::Error
///
/// ```
/// # mod hidden {
/// # #[allow(unused_imports)]
/// use tokio::fs::File;
/// # }
/// use tokio::io::AsyncRead;
/// use tokio_util::codec::{FramedRead, BytesCodec};
///
/// # enum File {}
/// # impl File {
/// #     async fn open(_name: &str) -> Result<impl AsyncRead, std::io::Error> {
/// #         use std::io::Cursor;
/// #         Ok(Cursor::new(vec![0, 1, 2, 3, 4, 5]))
/// #     }
/// # }
/// #
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() -> Result<(), std::io::Error> {
/// let my_async_read = File::open("filename.txt").await?;
/// let my_stream_of_bytes = FramedRead::new(my_async_read, BytesCodec::new());
/// # Ok(())
/// # }
/// ```
```

--------------------------------

### Example decode method with buffer management

Source: https://docs.rs/tokio-util/latest/src/tokio_util/codec/decoder.rs.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Shows a minimal implementation of the decode method for a custom codec. The method receives a mutable buffer and returns a decoded frame or None. The example includes a comment about buffer management, but the actual logic is omitted.

```rust
impl Decoder for MyCodec {
    // ...
    # type Item = BytesMut;
    # type Error = io::Error;

    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        // ...

```

--------------------------------

### Get number of items with len()

Source: https://docs.rs/tokio-util/latest/tokio_util/time/delay_queue/struct.DelayQueue.html

Returns the number of elements currently in the queue. The example creates an empty queue, asserts len is 0, inserts an item, and asserts len is 1.

```rust
use tokio_util::time::DelayQueue;
use std::time::Duration;

let mut delay_queue: DelayQueue<i32> = DelayQueue::with_capacity(10);
assert_eq!(delay_queue.len(), 0);
delay_queue.insert(3, Duration::from_secs(5));
assert_eq!(delay_queue.len(), 1);
```

--------------------------------

### Decoder implementation with buffer management strategy

Source: https://docs.rs/tokio-util/latest/src/tokio_util/codec/decoder.rs.html

Shows a Decoder implementation for MyCodec with a decode method. The example is from the 'Buffer management' section and demonstrates ensuring buffer capacity before returning. It is a no_run example with hidden lines for imports and type definitions.

```rust
impl Decoder for MyCodec {
    // ...
    # type Item = BytesMut;
    # type Error = io::Error;

    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        // ...
    }
}
```

--------------------------------

### Abort tasks matching a predicate with JoinMap::abort_all

Source: https://docs.rs/tokio-util/latest/src/tokio_util/task/join_map.rs.html?search=std%3A%3Avec

Shows how to abort all tasks whose keys satisfy a predicate. The example spawns four tasks and then aborts those with keys starting with "hello". The tasks are written to yield before completing so they can be aborted.

```rust
use tokio_util::task::JoinMap;

# // use the current thread rt so that spawned tasks don't
# // complete in the background before they can be aborted.
# #[tokio::main(flavor = "current_thread")]
# async fn main() {
let mut map = JoinMap::new();

map.spawn("hello world", async move {
    // ...
    # tokio::task::yield_now().await; // don't complete immediately, get aborted!
});
map.spawn("goodbye world", async move {
    // ...
    # tokio::task::yield_now().await; // don't complete immediately, get aborted!
});
map.spawn("hello san francisco", async move {
    // ...
    # tokio::task::yield_now().await; // don't complete immediately, get aborted!
});
map.spawn("goodbye universe", async move {
    // ...
    # tokio::task::yield_now().await; // don't complete immediately, get aborted!
});

map.abort_all(|key| key.starts_with("hello"));

while let Some((key, res)) = map.join_next().await {
    if key.starts_with("hello") {
        // The aborted tasks should complete with a cancelled `JoinError`.
        assert!(res.unwrap_err().is_cancelled());
    } else {
        // Other tasks should complete normally.
        assert!(res.is_ok());
    }
}
# }
```

--------------------------------

### Peek at next key in DelayQueue without removing it

Source: https://docs.rs/tokio-util/latest/src/tokio_util/time/delay_queue.rs.html

Shows how to use peek to get the Key that poll_expired will pull next without waiting. The example inserts three items with different delays and asserts that peek returns the key with the shortest delay (key2).

```rust
use tokio_util::time::DelayQueue;
use std::time::Duration;

# #[tokio::main(flavor = "current_thread")]
# async fn main() {
let mut delay_queue = DelayQueue::new();

let key1 = delay_queue.insert("foo", Duration::from_secs(10));
let key2 = delay_queue.insert("bar", Duration::from_secs(5));
let key3 = delay_queue.insert("baz", Duration::from_secs(15));

assert_eq!(delay_queue.peek().unwrap(), key2);
# }
```

--------------------------------

### TaskTracker Clone example

Source: https://docs.rs/tokio-util/latest/src/tokio_util/task/task_tracker.rs.html

Demonstrates that cloning a TaskTracker shares the same set of tasks, so spawns, close, and reopen are visible across clones. The example uses a tokio main function and shows how len, is_closed, and reopen behave on clones.

```rust
use tokio_util::task::TaskTracker;

#[tokio::main]
# async fn _hidden() {}
# #[tokio::main(flavor = "current_thread")]
async fn main() {
    let tracker = TaskTracker::new();
    let cloned = tracker.clone();

    // Spawns on `tracker` are visible in `cloned`.
    tracker.spawn(std::future::pending::<()>());
    assert_eq!(cloned.len(), 1);

    // Spawns on `cloned` are visible in `tracker`.
    cloned.spawn(std::future::pending::<()>());
    assert_eq!(tracker.len(), 2);

    // Calling `close` is visible to `cloned`.
    tracker.close();
    assert!(cloned.is_closed());

    // Calling `reopen` is visible to `tracker`.
    cloned.reopen();
    assert!(!tracker.is_closed());
}
```

--------------------------------

### Abort tasks matching a predicate with JoinMap::abort_matching

Source: https://docs.rs/tokio-util/latest/src/tokio_util/task/join_map.rs.html?search=

Shows how to abort all tasks whose keys match a predicate using JoinMap::abort_matching. The example spawns four tasks and aborts those whose keys start with "hello", then verifies that only the aborted tasks are cancelled.

```rust
use tokio_util::task::JoinMap;

# // use the current thread rt so that spawned tasks don't
# // complete in the background before they can be aborted.
# #[tokio::main(flavor = "current_thread")]
# async fn main() {
let mut map = JoinMap::new();

map.spawn("hello world", async move {
    // ...
    # tokio::task::yield_now().await; // don't complete immediately, get aborted!
});
map.spawn("goodbye world", async move {
    // ...
    # tokio::task::yield_now().await; // don't complete immediately, get aborted!
});
map.spawn("hello san francisco", async move {
    // ...
    # tokio::task::yield_now().await; // don't complete immediately, get aborted!
});
map.spawn("goodbye universe", async move {
    // ...
    # tokio::task::yield_now().await; // don't complete immediately, get aborted!
});

map.abort_matching(|key| key.starts_with("hello"));

let mut seen = map.join_next().await;

while let Some((key, res)) = seen {
    if key.starts_with("hello") {
        // The aborted task should complete with a cancelled `JoinError`.
        assert!(res.unwrap_err().is_cancelled());
    } else {
        // Other tasks should complete normally.
        assert!(res.is_ok());
    }
    seen = map.join_next().await;
}
# }
```

--------------------------------

### Abort tasks matching a predicate with JoinMap::abort_matching

Source: https://docs.rs/tokio-util/latest/tokio_util/task/struct.JoinMap.html

Shows how to abort all tasks whose keys satisfy a predicate using JoinMap::abort_matching. The example aborts tasks with keys starting with "goodbye" and verifies that aborted tasks yield cancelled JoinErrors while others complete normally.

```rust
use tokio_util::task::JoinMap;

let mut map = JoinMap::new();

map.spawn("hello world", async move {
    // ...
});
map.spawn("goodbye world", async move {
    // ...
});
map.spawn("hello san francisco", async move {
    // ...
});
map.spawn("goodbye universe", async move {
    // ...
});

// Abort all tasks whose keys begin with "goodbye"
map.abort_matching(|key| key.starts_with("goodbye"));

let mut seen = 0;
while let Some((key, res)) = map.join_next().await {
    seen += 1;
    if key.starts_with("goodbye") {
        // The aborted task should complete with a cancelled `JoinError`.
        assert!(res.unwrap_err().is_cancelled());
    } else {
        // Other tasks should complete normally.
        assert!(key.starts_with("hello"));
        assert!(res.is_ok());
    }
}

// All spawned tasks should have completed.
assert_eq!(seen, 4);
```

--------------------------------

### LocalPoolHandle example with spawn_local

Source: https://docs.rs/tokio-util/latest/src/tokio_util/task/spawn_pinned.rs.html

Shows how to create a `LocalPoolHandle` with 5 threads and spawn a `!Send` future using `spawn_pinned`. Inside the future, `spawn_local` is used to spawn another `!Send` task on the same thread. The example is wrapped in `#[cfg(not(target_family = "wasm"))]` and uses `#[tokio::main(flavor = "current_thread")]`.

```rust
use std::rc::Rc;
use tokio::task;
use tokio_util::task::LocalPoolHandle;

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let pool = LocalPoolHandle::new(5);

    let output = pool.spawn_pinned(|| {
        // `data` is !Send + !Sync
        let data = Rc::new("local data");
        let data_clone = data.clone();

        async move {
            task::spawn_local(async move {
                println!("{}", data_clone);
            });

            data.to_string()
        }
    }).await.unwrap();
    println!("output: {}", output);
}
```

--------------------------------

### TaskTracker::new

Source: https://docs.rs/tokio-util/latest/src/tokio_util/task/task_tracker.rs.html

Creates a new TaskTracker that starts out open.

```APIDOC
## TaskTracker::new

### Description
Creates a new `TaskTracker`. The `TaskTracker` will start out as open.

### Method
`pub fn new() -> Self`

### Parameters
None.

### Returns
A new `TaskTracker` instance.

### Example
```rust
let tracker = TaskTracker::new();
```
```

--------------------------------

### Aborting tasks by predicate with JoinMap::abort_matching

Source: https://docs.rs/tokio-util/latest/tokio_util/task/struct.JoinMap.html?search=

Shows how to abort all tasks whose keys match a predicate using JoinMap::abort_matching. The example spawns four tasks, aborts those with keys starting with "goodbye", and verifies that aborted tasks yield cancelled JoinErrors while others complete successfully.

```rust
use tokio_util::task::JoinMap;

let mut map = JoinMap::new();

map.spawn("hello world", async move {
    // ...
});
map.spawn("goodbye world", async move {
    // ...
});
map.spawn("hello san francisco", async move {
    // ...
});
map.spawn("goodbye universe", async move {
    // ...
});

// Abort all tasks whose keys begin with "goodbye"
map.abort_matching(|key| key.starts_with("goodbye"));

let mut seen = 0;
while let Some((key, res)) = map.join_next().await {
    seen += 1;
    if key.starts_with("goodbye") {
        // The aborted task should complete with a cancelled `JoinError`.
        assert!(res.unwrap_err().is_cancelled());
    } else {
        // Other tasks should complete normally.
        assert!(key.starts_with("hello"));
        assert!(res.is_ok());
    }
}

// All spawned tasks should have completed.
assert_eq!(seen, 4);

```

--------------------------------

### join_all

Source: https://docs.rs/tokio-util/latest/src/tokio_util/task/join_queue.rs.html?search=std%3A%3Avec

Awaits all tasks and returns their results in spawn order, panicking on any error.

```APIDOC
## join_all

### Description
Awaits the completion of all tasks in the `JoinQueue`, returning a vector of their results. The results are stored in the order the tasks were spawned, not the order they completed. This is a convenience method equivalent to calling `join_next` in a loop. If any task fails with a `JoinError`, this call panics and all remaining tasks are cancelled. To handle errors differently, manually call `join_next` in a loop.

### Method
`pub async fn join_all(mut self) -> Vec<T>`

### Parameters
None.

### Returns
- `Vec<T>` - A vector of the results of all tasks, in spawn order.

### Cancel Safety
This method is not cancel safe because it calls `join_next` in a loop. For cancel safety, manually call `join_next` in a loop with a `Vec` accumulator.

### Notes
- Panics if any task fails with a `JoinError` (including panics and aborts).
- See also: [`join_next`], [`JoinError::id`].
```

--------------------------------

### Abort tasks matching a predicate with abort_matching

Source: https://docs.rs/tokio-util/latest/src/tokio_util/task/join_map.rs.html?search=u32+-%3E+bool

Aborts all tasks whose keys satisfy the given predicate. The example spawns four tasks, aborts those with keys starting with 'goodbye', and then joins all tasks, asserting that aborted tasks return a cancelled JoinError and others complete normally. The method iterates without removing entries so aborted keys are still returned by join_next.

```rust
/// });
///
/// // Abort all tasks whose keys begin with "goodbye"
/// map.abort_matching(|key| key.starts_with("goodbye"));
///
/// let mut seen = 0;
/// while let Some((key, res)) = map.join_next().await {
///     seen += 1;
///     if key.starts_with("goodbye") {
///         // The aborted task should complete with a cancelled `JoinError`.
///         assert!(res.unwrap_err().is_cancelled());
///     } else {
///         // Other tasks should complete normally.
///         assert!(key.starts_with("hello"));
///         assert!(res.is_ok());
///     }
/// }
///
/// // All spawned tasks should have completed.
/// assert_eq!(seen, 4);
/// # }
/// ```
```

--------------------------------

### DelayQueue Usage Example

Source: https://docs.rs/tokio-util/latest/tokio_util/time/struct.DelayQueue.html?search=u32+-%3E+bool

Example demonstrating how to use DelayQueue to manage cache entries with expiration.

```APIDOC
## Usage Example

### Description
Shows how to use `DelayQueue` to manage cache entries with a TTL.

### Code
```rust
use tokio_util::time::{DelayQueue, delay_queue};

use std::collections::HashMap;
use std::task::{ready, Context, Poll};
use std::time::Duration;

struct Cache {
    entries: HashMap<CacheKey, (Value, delay_queue::Key)>,
    expirations: DelayQueue<CacheKey>,
}

const TTL_SECS: u64 = 30;

impl Cache {
    fn insert(&mut self, key: CacheKey, value: Value) {
        let delay = self.expirations
            .insert(key.clone(), Duration::from_secs(TTL_SECS));

        self.entries.insert(key, (value, delay));
    }

    fn get(&self, key: &CacheKey) -> Option<&Value> {
        self.entries.get(key)
            .map(|&(ref v, _)| v)
    }

    fn remove(&mut self, key: &CacheKey) {
        if let Some((_, cache_key)) = self.entries.remove(key) {
            self.expirations.remove(&cache_key);
        }
    }

    fn poll_purge(&mut self, cx: &mut Context<'_>) -> Poll<()> {
        while let Some(entry) = ready!(self.expirations.poll_expired(cx)) {
            self.entries.remove(entry.get_ref());
        }

        Poll::Ready(())
    }
}
```
```

--------------------------------

### fn compat_write(self) -> Compat<Self>

Source: https://docs.rs/tokio-util/latest/tokio_util/compat/trait.TokioAsyncWriteCompatExt.html

Wraps self with a compatibility layer that implements futures_io::AsyncWrite.

```APIDOC
## fn compat_write(self) -> Compat<Self>

### Description
Wraps `self` with a compatibility layer that implements `futures_io::AsyncWrite`.

### Method
`compat_write`

### Signature
```rust
fn compat_write(self) -> Compat<Self>
where
    Self: Sized,
```

### Parameters
- `self` (Self) - Required - The value to wrap.

### Return Value
- `Compat<Self>` - A compatibility layer that implements `futures_io::AsyncWrite`.

### Availability
This method is available on **crate feature `compat`** only.

### Implementors
- `impl<T: AsyncWrite> TokioAsyncWriteCompatExt for T`

### Dyn Compatibility
This trait **is** dyn compatible.
```

--------------------------------

### poll_write_buf example: writing Cursor to File

Source: https://docs.rs/tokio-util/latest/src/tokio_util/util/poll_buf.rs.html?search=

Example usage of poll_write_buf to write the contents of a Cursor<&[u8]> to a File. It loops until the buffer has no remaining bytes, using poll_fn to poll the write operation. The example is marked as no_run because it creates a file.

```rust
use tokio_util::io::poll_write_buf;
use tokio::io;
use tokio::fs::File;

use bytes::Buf;
use std::future::poll_fn;
use std::io::Cursor;
use std::pin::Pin;

#[tokio::main]
async fn main() -> io::Result<()> {
    let mut file = File::create("foo.txt").await?;
    let mut buf = Cursor::new(b"data to write");

    // Loop until the entire contents of the buffer are written to
    // the file.
    while buf.has_remaining() {
        poll_fn(|cx| poll_write_buf(Pin::new(&mut file), cx, &mut buf)).await?;
    }

    Ok(())
}
```

--------------------------------

### Example: Using RuntimeExt::wrap to run a sleep future on a runtime without timers

Source: https://docs.rs/tokio-util/latest/src/tokio_util/context.rs.html

This example creates two runtimes, one with timers enabled and one without. It wraps a sleep future in the context of the timer-enabled runtime and executes it on the timer-disabled runtime, demonstrating how to use the runtime context to enable time features.

```rust
# #[cfg(not(target_family = "wasm"))]
# {
use tokio::time::{sleep, Duration};
use tokio_util::context::RuntimeExt;

// This runtime has timers enabled.
let rt = tokio::runtime::Builder::new_multi_thread()
    .enable_all()
    .build()
    .unwrap();

// This runtime has timers disabled.
let rt2 = tokio::runtime::Builder::new_multi_thread()
    .build()
    .unwrap();

// Wrap the sleep future in the context of rt.
let fut = rt.wrap(async { sleep(Duration::from_millis(2)).await });

// Execute the future on rt2.
rt2.block_on(fut);
# }
```

--------------------------------

### CancellationToken::clone

Source: https://docs.rs/tokio-util/latest/src/tokio_util/sync/cancellation_token.rs.html

Creates a clone of the token that will get cancelled whenever the current token gets cancelled, and vice versa.

```APIDOC
## CancellationToken::clone

### Description
Creates a clone of the token that will get cancelled whenever the current token gets cancelled, and vice versa.

### Method
clone

### Signature
fn clone(&self) -> Self

### Parameters
- **self** (&self) - The token to clone.

### Returns
- **Self** - A new token that shares the same cancellation state.
```

--------------------------------

### Example: Using CancellationToken with tokio::select!

Source: https://docs.rs/tokio-util/latest/src/tokio_util/sync/cancellation_token.rs.html?search=

Shows how to create a CancellationToken, clone it for a spawned task, and cancel it after a delay. The spawned task waits for either cancellation or a long sleep, returning 5 on cancellation. The example asserts the result is 5.

```rust
use tokio::select;
use tokio_util::sync::CancellationToken;

#[tokio::main]
async fn main() {
    let token = CancellationToken::new();
    let cloned_token = token.clone();

    let join_handle = tokio::spawn(async move {
        // Wait for either cancellation or a very long time
        select! {
            _ = cloned_token.cancelled() => {
                // The token was cancelled
                5
            }
            _ = tokio::time::sleep(std::time::Duration::from_secs(9999)) => {
                99
            }
        }
    });

    tokio::spawn(async move {
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        token.cancel();
    });

    assert_eq!(5, join_handle.await.unwrap());
}
```

--------------------------------

### CancellationToken::clone

Source: https://docs.rs/tokio-util/latest/src/tokio_util/sync/cancellation_token.rs.html?search=

Creates a clone of the CancellationToken which will get cancelled whenever the current token gets cancelled, and vice versa.

```APIDOC
## CancellationToken::clone

### Description
Creates a clone of the CancellationToken which will get cancelled whenever the current token gets cancelled, and vice versa.

### Method
`pub fn clone(&self) -> CancellationToken`

### Parameters
- **self** (`&self`) - The cancellation token.

### Returns
- **CancellationToken** - A new token that is linked to the original.

### Example
```rust
use tokio_util::sync::CancellationToken;
let token = CancellationToken::new();
let cloned = token.clone();
```
```

--------------------------------

### poll_read_buf example: reading from StreamReader into BytesMut

Source: https://docs.rs/tokio-util/latest/src/tokio_util/util/poll_buf.rs.html?search=

Example usage of poll_read_buf to read data from a StreamReader into a BytesMut. It uses poll_fn to poll the read operation in a loop until EOF. The example asserts that at least one read occurs and the buffer contains the expected bytes.

```rust
use bytes::{Bytes, BytesMut};
use tokio_stream as stream;
use tokio::io::Result;
use tokio_util::io::{StreamReader, poll_read_buf};
use std::future::poll_fn;
use std::pin::Pin;
# #[tokio::main(flavor = "current_thread")]
# async fn main() -> std::io::Result<()> {

// Create a reader from an iterator. This particular reader will always be
// ready.
let mut read = StreamReader::new(stream::iter(vec![Result::Ok(Bytes::from_static(&[0, 1, 2, 3]))]));

let mut buf = BytesMut::new();
let mut reads = 0;

loop {
    reads += 1;
    let n = poll_fn(|cx| poll_read_buf(Pin::new(&mut read), cx, &mut buf)).await?;

    if n == 0 {
        break;
    }
}

// one or more reads might be necessary.
assert!(reads >= 1);
assert_eq!(&buf[..], &[0, 1, 2, 3]);
# Ok(())
# }
```

--------------------------------

### new

Source: https://docs.rs/tokio-util/latest/tokio_util/io/simplex/index.html?search=std%3A%3Avec

Creates a simplex channel.

```APIDOC
## new

### Description
Creates a simplex channel.

### Function
`new`

### Parameters
None documented.

### Returns
A simplex channel (Sender and Receiver pair).
```

--------------------------------

### TaskTracker graceful shutdown example

Source: https://docs.rs/tokio-util/latest/src/tokio_util/task/task_tracker.rs.html?search=std%3A%3Avec

Example showing how to use TaskTracker with a graceful shutdown pattern. It spawns tasks, then a background task sends a shutdown signal after 2 seconds, and the main task waits for all tasks to exit.

```rust
use std::time::Duration;
use tokio::time;
use tokio_util::task::TaskTracker;

#[tokio::main]
async fn main() {
    let tracker = TaskTracker::new();

    for i in 0..10 {
        tracker.spawn(async move {
            // Code to run in the background task.
            time::sleep(Duration::from_secs(1)).await;
            println!("Task {} is running.", i);
        });
    }

    // Wait for all tasks to exit.
    tracker.wait().await;

    println!("All tasks have exited now.");
}
```
Tokens:
Raw