Futures Crate

Recipe	Crates	Categories
Selecting Futures
Joining Futures
Map, then, either, flatten
futures-util

The futures⮳ crate provides a number of core abstractions for writing asynchronous code.

In most cases, you will use this crate directly only when writing async code intended to work for multiple runtimes. Otherwise, use the utilities provided by the ecosystem of your choice - Tokio for example.

Selecting Futures

futures::future::Select⮳ polls multiple futures and streams simultaneously, executing the branch for the future that finishes first. If multiple futures are ready, one will be pseudo-randomly selected at runtime.

use futures::{
    future::FutureExt, // Provides the `.fuse()` method for futures.
    pin_mut,
    select,
};

// Define two asynchronous tasks.
async fn task_one() {
    // ...
}
async fn task_two() {
    // ...
}

// Define an asynchronous function that races two tasks.
async fn race_tasks() {
    // Fuse the two tasks so that they can be used with `select!`.
    let t1 = task_one().fuse();
    let t2 = task_two().fuse();

    // Pin the futures to the stack using `pin_mut!`.
    pin_mut!(t1, t2);

    // Use `select!` to race the two tasks.
    select! {
        () = t1 => println!("task one completed first"),
        () = t2 => println!("task two completed first"),
    }
}

#[tokio::main]
async fn main() {
    race_tasks().await;
}

Joining Futures

use futures::join;

/// Print the input value and return it.
async fn foo(i: u32) -> u32 {
    println!("{i}");
    i
}

#[tokio::main]
async fn main() {
    // The `join!` macro polls multiple futures simultaneously, returning
    // a tuple of all results once complete.
    assert_eq!(join!(foo(1), foo(2)), (1, 2));
    // `join!` is variadic, so you can pass any number of futures

    // `join_all` create a future which represents a collection of the
    // outputs of the futures given.
    let futures = vec![foo(1), foo(2), foo(3)];
    assert_eq!(futures::future::join_all(futures).await, [1, 2, 3]);
}

Map, then, either, flatten

The futures⮳ crate provides an extension trait that provides a variety of convenient adapters.

//! # Futures
//!
//! This example demonstrates various operations on futures using the `futures`
//! crate.
//!
//! It showcases how to:
//! - Map the output of a future to a different type.
//! - Chain computations onto a future using `then`.
//! - Use conditional `Either` futures.
//! - Flatten nested futures.
use anyhow::Result;
use futures::future::FutureExt;

#[tokio::main]
async fn main() -> Result<()> {
    // Create a future that resolves to the value 1.
    let future_of_1 = async { 1 };

    // Map this future's output to a (possibly) different type, returning
    // a new future of the resulting type.
    let new_future = future_of_1.map(|x| x + 3);

    // Chain on a computation for when a future finished, passing the
    // result of the future to the provided closure f.
    let future_of_7 = new_future.then(|x| async move { x + 3 });
    let seven = future_of_7.await;
    println!("{}", seven);
    assert_eq!(seven, 7);

    // Conditional `Either` future
    let x = 6;
    let future = if x > 10 {
        async { true }.left_future()
    } else {
        async { false }.right_future()
    };
    let not_true: bool = future.await;
    assert!(!not_true);

    // Flatten nested futures:
    let nested_future = async { async { 1 } };
    let future = nested_future.flatten();
    let flat = future.await;
    println!("{flat}");
    assert_eq!(flat, 1);
    Ok(())
}

futures-util

Common utilities and extension traits for the futures-rs⮳ library. Extensions to Rust's Future and Stream traits. Combinators and utilities for working with Futures⮳, Streams, Sinks, and the AsyncRead and AsyncWrite traits.

use std::time::Duration;

use futures::future;
use futures_util::future::FutureExt;
use futures_util::future::TryFutureExt;
use futures_util::pin_mut;
use tokio::time;

// This example demonstrates how to use `futures-util` to work with
// asynchronous code.

// Add these dependencies to your Cargo.toml:
// [dependencies]
// futures = "0.3"
// futures-util = "0.3"
// tokio = { version = "1", features = ["full"] }

/// Simulated async function that returns a `Result`:
async fn fetch_data(id: u32) -> Result<String, String> {
    // Simulate network delay
    time::sleep(Duration::from_millis(100)).await;
    if id % 3 == 0 {
        Err(format!("Error fetching data for id {}", id))
    } else {
        Ok(format!("Data for id {}", id))
    }
}

#[tokio::main]
async fn main() {
    println!("\n===== Combining Futures =====");
    // Demonstrates:
    // `select` for racing futures,
    // `join_all` for waiting on multiple futures,
    // `try_join` for handling errors in combined futures.

    // `ready` creates a future that is immediately ready with a value.
    let future1 =
        time::sleep(Duration::from_millis(100)).then(|_| future::ready(1));
    let future2 =
        time::sleep(Duration::from_millis(50)).then(|_| future::ready(2));

    // `select` requires `Future` + `Unpin` bounds.
    pin_mut!(future1);
    pin_mut!(future2);

    // Select between multiple futures:
    match future::select(future1, future2).await {
        future::Either::Left((val, _)) => {
            println!("Future 1 completed first with: {}", val)
        }
        future::Either::Right((val, _)) => {
            println!("Future 2 completed first with: {}", val)
        }
    }

    // Join futures:
    let futures = vec![fetch_data(1), fetch_data(2)];
    // The future returned by `join_all` will drive execution for all of its
    // underlying futures, collecting the results into a destination Vec<T>
    // in the same order as they were provided.
    let results: Vec<Result<String, String>> = future::join_all(futures).await;
    println!("Join results: {:?}", results);

    // `try_join`:
    // If successful, the returned future will finish with a tuple of both
    // results. If unsuccessful, it will complete with the first error
    // encountered.
    let results = future::try_join(fetch_data(1), fetch_data(4)).await;
    println!("try_join results: {:?}", results);

    println!("===== Future Extensions =====");
    // Demonstrates:
    // `map` to transform future outputs,
    // `and_then` for chaining asynchronous operations,
    // `or_else` for error handling.

    // Use `map` to transform the output of a Future
    let future = fetch_data(1).map(|result| match result {
        Ok(data) => Ok(format!("Processed: {}", data)),
        Err(e) => Err(e),
    });
    println!("Map result: {:?}", future.await);

    // Use `and_then` for chaining futures:
    let future = fetch_data(2).and_then(|data| async move {
        // Simulate additional processing
        time::sleep(Duration::from_millis(50)).await;
        Ok(format!("Enhanced: {}", data))
    });
    println!("AndThen result: {:?}", future.await);

    // Use `or_else` for error handling:
    let future = fetch_data(3).or_else(|err| async move {
        println!("Recovering from error: {}", err);
        // Return a fallback value
        Ok::<_, String>(String::from("Fallback data"))
    });
    println!("OrElse result: {:?}", future.await);
}

Related Topics

• Tokio.

See Also