Message Passing

When your program has multiple tasks running at the same time, they sometimes need to talk to each other. Channels are how you do that, one task sends a message, another receives it. Think of it like a pipe: you put something in one end, and it comes out the other.

Bounded Channels

A bounded channel has a limit on how many messages it can hold at once. If the channel is full, the sender has to wait until there is room before it can send another message.

Think of it like a physical mailbox. It can only hold a fixed number of letters. If it's full, the postman has to wait until someone clears it out before dropping in another letter.

In this example, two book stores send books through a channel with a capacity of 5. A shelf collects whatever comes through.

use std::io;
use tokio::sync::mpsc::channel;

struct Book {
    title: &'static str,
}

impl Book {
    fn new(title: &'static str) -> Self {
        Self { title }
    }
}

#[tokio::main]
async fn main() -> io::Result<()> {
    let (book_sender, mut book_receiver) = channel(5);

    // Each store gets its own copy of the sender.
    let store_one_book_sender = book_sender.clone();
    let book_store_one = tokio::task::spawn(async move {
        if let Err(err) = store_one_book_sender
            .send(Book::new("Shawshank Redemption"))
            .await
        {
            eprintln!("Failed to send book from store one: {}", err);
        }
    });

    let book_store_two = tokio::task::spawn(async move {
        if let Err(err) = book_sender.send(Book::new("Secret Recipe")).await {
            eprintln!("Failed to send book from store two: {}", err);
        }
    });

    let mut shelf: Vec<Book> = Vec::new();
    // Collect every book that arrives until both stores are done sending.
    while let Some(new_book) = book_receiver.recv().await {
        shelf.push(new_book);
    }

    book_store_one.await?;
    book_store_two.await?;

    for book in &shelf {
        println!("Title: {}", book.title);
    }

    Ok(())
}

Add tokio to Cargo.toml with the macros and sync features enabled.

[dependencies]
tokio = { version = "*", features = ["macros", "sync"] }

Unbounded Channels

An unbounded channel has no limit on how many messages it can hold. The sender never has to wait, it can always drop a message in, no matter how many are already sitting there.

Think of it like a digital inbox. It just keeps growing as new messages arrive. There's no cap, but if messages pile up faster than they're being read, your program will use more and more memory. Sending on an unbounded channel will always succeed as long as the receiving end is still open. If the receiver is slow, messages simply queue up and wait.

In this example, two people send messages through a channel. An inbox collects whatever comes through.

use std::io;
use tokio::sync::mpsc::unbounded_channel;

struct Message {
    from: &'static str,
    text: &'static str,
}

impl Message {
    fn new(from: &'static str, text: &'static str) -> Self {
        Self { from, text }
    }
}

#[tokio::main]
async fn main() -> io::Result<()> {
    let (message_sender, mut message_receiver) = unbounded_channel();

    let alice_message_sender = message_sender.clone();
    let person_one = tokio::task::spawn(async move {
        if let Err(err) = alice_message_sender.send(Message::new("Alice", "Meeting postponed")) {
            eprintln!("Failed to send message from Alice: {}", err);
        }
    });

    let person_two = tokio::task::spawn(async move {
        if let Err(err) = message_sender.send(Message::new("Bob", "Secret Leaked")) {
            eprintln!("Failed to send message from Bob: {}", err);
        }
    });

    let mut inbox: Vec<Message> = Vec::new();
    while let Some(new_book) = message_receiver.recv().await {
        inbox.push(new_book);
    }

    person_one.await?;
    person_two.await?;

    for msg in &inbox {
        println!("{} says: {}", msg.from, msg.text);
    }

    Ok(())
}

Add tokio to Cargo.toml with the macros and sync features enabled.

[dependencies]
tokio = { version = "*", features = ["macros", "sync"] }