oneshot, broadcast, and watch

mpsc channels move a stream of messages from many senders to one receiver. Three other channels in tokio::sync cover different shapes: one reply to one caller, one message to many receivers, and one shared value that changes over time. After this lesson you can pick the right channel for a job and handle the errors each one reports.

Channel	Shape	Retains	Main calls
`oneshot`	one sender, one receiver, one value	the single value	`channel`, `send`, `.await` on the receiver, `try_recv`, `closed`
`broadcast`	many senders, many receivers, every receiver sees every message	a ring buffer of the last N messages	`channel`, `send`, `subscribe`, `recv`, `receiver_count`
`watch`	many senders, many receivers, receivers see the latest value	one value	`channel`, `send`, `borrow`, `borrow_and_update`, `changed`, `send_modify`, `send_if_modified`

All three are in the sync feature of Tokio (part of full). Their send calls are not async, so they can be called from any code, including plain threads.

Example 1: The smallest oneshot

A oneshot channel carries exactly one value from one sender to one receiver. oneshot::channel() returns a Sender and a Receiver. The receiver is a future: awaiting it waits for the value.

Rust

use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let (tx, rx) = oneshot::channel();

    tokio::spawn(async move {
        tx.send(42).unwrap();
    });

    let value = rx.await.unwrap();
    println!("received {value}");
}

Output

received 42

The task sends 42 and the main task receives it. Both send and .await return a Result, and Examples 2 to 4 show when each one is an error.

Example 2: send fails when the receiver is gone

Sender::send takes self, so a sender works once. It is not async because the channel never waits for space. If the receiver has been dropped, send returns Err and hands the value back to the caller.

Rust

use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let (tx, rx) = oneshot::channel::<String>();
    drop(rx);

    println!("is_closed: {}", tx.is_closed());
    match tx.send(String::from("report")) {
        Ok(()) => println!("delivered"),
        Err(value) => println!("returned to sender: {value}"),
    }
}

Output

is_closed: true
returned to sender: report

Returning the value lets the caller reuse it or log it instead of losing it.

Example 3: a sender cannot be used twice

send consumes the sender, so the compiler rejects a second send. The type system enforces the "one value" rule.

Rust✗ does not compile · E0382

use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let (tx, _rx) = oneshot::channel();
    tx.send(1).unwrap();
    tx.send(2).unwrap();
}

Output

error[E0382]: use of moved value: `tx`
   --> src/main.rs:7:5
    |
  5 |     let (tx, _rx) = oneshot::channel();
    |          -- move occurs because `tx` has type `tokio::sync::oneshot::Sender<i32>`, which does not implement the `Copy` trait
  6 |     tx.send(1).unwrap();
    |        ------- `tx` moved due to this method call
  7 |     tx.send(2).unwrap();
    |     ^^ value used here after move
    |
note: `tokio::sync::oneshot::Sender::<T>::send` takes ownership of the receiver `self`, which moves `tx`
   --> ~/.cargo/registry/src/.../tokio-1.53.1/src/sync/oneshot.rs:622:21
    |
622 |     pub fn send(mut self, t: T) -> Result<(), T> {
    |                     ^^^^

Example 4: a dropped sender ends the wait

If the Sender is dropped without sending, awaiting the receiver returns Err(RecvError). This makes a crashed or cancelled producer visible to the consumer instead of leaving it waiting forever.

Rust

use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let (tx, rx) = oneshot::channel::<u32>();

    tokio::spawn(async move {
        // The task ends without sending. `tx` is dropped here.
        let _keep = tx;
    });

    match rx.await {
        Ok(value) => println!("value: {value}"),
        Err(error) => println!("error: {error} / {error:?}"),
    }
}

Output

error: channel closed / RecvError(())

RecvError has no fields and the same Display text as the other channel errors: channel closed.

Example 5: try_recv without waiting

Receiver::try_recv checks for a value without awaiting. It returns Err(TryRecvError::Empty) while the sender is alive and has not sent, and Err(TryRecvError::Closed) when the sender was dropped without a value or the value was already taken.

Rust

use tokio::sync::oneshot;
use tokio::sync::oneshot::error::TryRecvError;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = oneshot::channel();

    println!("before send: {:?}", rx.try_recv());
    tx.send("ready").unwrap();
    println!("after send: {:?}", rx.try_recv());
    println!("after taking the value: {:?}", rx.try_recv());

    let (tx2, mut rx2) = oneshot::channel::<u8>();
    drop(tx2);
    let closed = matches!(rx2.try_recv(), Err(TryRecvError::Closed));
    println!("sender dropped: closed = {closed}");
}

Output

before send: Err(Empty)
after send: Ok("ready")
after taking the value: Err(Closed)
sender dropped: closed = true

Awaiting the receiver takes it by value, so try_recv needs mut rx and keeps the receiver usable.

Example 6: request and response

The most common use of oneshot is a reply channel. The caller creates a oneshot, puts the Sender inside the request, and sends the request over an mpsc channel. The service answers on the oneshot. Each request carries its own reply path, so replies cannot be mixed up.

Rust

use tokio::sync::{mpsc, oneshot};

enum Request {
    Square { n: u64, reply: oneshot::Sender<u64> },
    Name { reply: oneshot::Sender<&'static str> },
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<Request>(8);

    let service = tokio::spawn(async move {
        while let Some(request) = rx.recv().await {
            match request {
                Request::Square { n, reply } => {
                    let _ = reply.send(n * n);
                }
                Request::Name { reply } => {
                    let _ = reply.send("squarer");
                }
            }
        }
        println!("service stopped");
    });

    let (reply_tx, reply_rx) = oneshot::channel();
    tx.send(Request::Square { n: 12, reply: reply_tx }).await.unwrap();
    println!("12 squared = {}", reply_rx.await.unwrap());

    let (reply_tx, reply_rx) = oneshot::channel();
    tx.send(Request::Name { reply: reply_tx }).await.unwrap();
    println!("name = {}", reply_rx.await.unwrap());

    drop(tx);
    service.await.unwrap();
}

Output

12 squared = 144
name = squarer
service stopped

The service uses let _ = reply.send(...) because the caller may have stopped waiting. That is not an error for the service. Dropping tx closes the request channel, recv returns None, and the service ends. The Actors, Worker Pools, and Pipelines lesson builds this into a full pattern.

Example 7: detect that the caller gave up

Sender::closed().await completes when the receiver is dropped. A worker can race it against the real work and stop early. The example uses Tokio's paused clock (time::pause, taught in Time: sleep, timeout, and interval) so the ten-second job finishes instantly and the output does not depend on real time. time::pause needs the current-thread runtime, so the example selects it with flavor = "current_thread".

Rust

use std::time::Duration;
use tokio::sync::oneshot;
use tokio::time::{self, sleep};

async fn slow_job(mut reply: oneshot::Sender<&'static str>, label: &'static str) {
    tokio::select! {
        _ = reply.closed() => println!("{label}: caller left, work skipped"),
        _ = sleep(Duration::from_secs(10)) => {
            let sent = reply.send("done").is_ok();
            println!("{label}: work finished, delivered = {sent}");
        }
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    time::pause();

    let (tx, rx) = oneshot::channel();
    let job = tokio::spawn(slow_job(tx, "job A"));
    sleep(Duration::from_secs(1)).await;
    drop(rx);
    job.await.unwrap();

    let (tx, rx) = oneshot::channel();
    let job = tokio::spawn(slow_job(tx, "job B"));
    println!("job B reply: {}", rx.await.unwrap());
    job.await.unwrap();
}

Output

job A: caller left, work skipped
job B: work finished, delivered = true
job B reply: done

Job A stops after one virtual second because its receiver was dropped. Job B runs to the end and its reply arrives.

Example 8: a timeout around a reply

A oneshot receiver is a future, so tokio::time::timeout can wrap it. When the timeout expires, timeout drops the receiver, and the late send then fails.

Rust

use std::time::Duration;
use tokio::sync::oneshot;
use tokio::time::{self, sleep, timeout};

#[tokio::main(flavor = "current_thread")]
async fn main() {
    time::pause();

    let (tx, rx) = oneshot::channel();
    let slow = tokio::spawn(async move {
        sleep(Duration::from_secs(5)).await;
        tx.send("late answer")
    });

    match timeout(Duration::from_secs(2), rx).await {
        Ok(Ok(value)) => println!("answer: {value}"),
        Ok(Err(_)) => println!("sender dropped"),
        Err(elapsed) => println!("gave up: {elapsed}"),
    }

    let late_send = slow.await.unwrap();
    println!("late send accepted: {}", late_send.is_ok());
}

Output

gave up: deadline has elapsed
late send accepted: false

Example 9: broadcast delivers every message to every receiver

A broadcast channel has many receivers, and each receiver gets its own copy of every message. broadcast::channel(capacity) returns the first Sender and Receiver. Sender::subscribe creates more receivers. The message type must implement Clone, because each receiver gets a clone. send returns the number of receivers that were subscribed at that moment.

Rust

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (tx, mut rx1) = broadcast::channel(16);
    let mut rx2 = tx.subscribe();

    let receivers = tx.send("hello").unwrap();
    println!("send reached {receivers} receivers");
    tx.send("world").unwrap();

    println!("rx1: {} {}", rx1.recv().await.unwrap(), rx1.recv().await.unwrap());
    println!("rx2: {} {}", rx2.recv().await.unwrap(), rx2.recv().await.unwrap());
}

Output

send reached 2 receivers
rx1: hello world
rx2: hello world

The value is stored once inside the channel and cloned for each receiver when it reads it.

A receiver created by subscribe starts at the current end of the channel. It does not see earlier messages, even if the buffer still holds them. receiver_count reports how many receivers exist. send returns Err with the value when there are none.

Rust

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (tx, rx) = broadcast::channel(16);
    println!("receivers: {}", tx.receiver_count());

    tx.send("first").unwrap();

    let mut late = tx.subscribe();
    println!("receivers: {}", tx.receiver_count());
    tx.send("second").unwrap();
    println!("late receiver sees: {}", late.recv().await.unwrap());

    drop(rx);
    drop(late);
    println!("receivers: {}", tx.receiver_count());
    match tx.send("nobody") {
        Ok(_) => println!("delivered"),
        Err(error) => println!("send failed: {error}, value = {}", error.0),
    }

    let mut again = tx.subscribe();
    tx.send("third").unwrap();
    println!("new receiver sees: {}", again.recv().await.unwrap());
}

Output

receivers: 1
receivers: 2
late receiver sees: second
receivers: 0
send failed: channel closed, value = nobody
new receiver sees: third

An Err from send does not close the channel. A later subscribe makes sending work again. Sender::new(capacity) creates a sender with zero receivers, which suits a channel that receivers join later.

Example 11: a slow receiver lags

The channel keeps at most capacity messages. When a send finds the buffer full, it overwrites the oldest message. A receiver that had not read it gets Err(RecvError::Lagged(n)) on its next recv, where n is the number of messages it missed. The receiver stays subscribed, and the next recv returns the oldest message still in the buffer.

The buffer length is capacity rounded up to the next power of two (channel(3) holds 4). This example uses a power of two so the numbers are exact.

Rust

use tokio::sync::broadcast;
use tokio::sync::broadcast::error::RecvError;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = broadcast::channel(4);

    for n in 1..=6 {
        tx.send(n).unwrap();
    }

    loop {
        match rx.recv().await {
            Ok(n) => println!("got {n}"),
            Err(RecvError::Lagged(missed)) => println!("lagged, missed {missed}"),
            Err(RecvError::Closed) => break,
        }
        if rx.is_empty() {
            break;
        }
    }
}

Output

lagged, missed 2
got 3
got 4
got 5
got 6

Messages 1 and 2 were overwritten by 5 and 6. The receiver reports 2 missed messages and then reads 3, 4, 5, and 6 in order. A receiver decides whether missing messages is acceptable (metrics, live prices) or fatal (it can stop the task).

Example 12: closing and draining

When every Sender is dropped, the channel closes. Receivers still read the messages that remain in the buffer, and after the last one recv returns Err(RecvError::Closed). This is the normal way a receiver loop ends.

Rust

use tokio::sync::broadcast;
use tokio::sync::broadcast::error::{RecvError, TryRecvError};

#[tokio::main]
async fn main() {
    let (tx, mut rx) = broadcast::channel(8);
    println!("empty: {:?}", rx.try_recv());

    tx.send("a").unwrap();
    tx.send("b").unwrap();
    drop(tx);

    loop {
        match rx.recv().await {
            Ok(message) => println!("message {message}"),
            Err(RecvError::Closed) => {
                println!("closed");
                break;
            }
            Err(RecvError::Lagged(_)) => unreachable!("capacity is large enough"),
        }
    }
    println!("try_recv after close: {:?}", rx.try_recv());
    assert_eq!(rx.try_recv(), Err(TryRecvError::Closed));
}

Output

empty: Err(Empty)
message a
message b
closed
try_recv after close: Err(Closed)

try_recv returns Empty, Closed, or Lagged(n) without waiting.

Example 13: fan-out to several tasks

Three worker tasks each subscribe to a broadcast channel of events. Each task collects every event until the channel closes. All receivers are created before the first send, so each one sees the full list.

Rust

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (tx, _) = broadcast::channel::<String>(16);

    let mut handles = Vec::new();
    for id in 1..=3 {
        let mut rx = tx.subscribe();
        handles.push(tokio::spawn(async move {
            let mut seen = Vec::new();
            while let Ok(event) = rx.recv().await {
                seen.push(event);
            }
            (id, seen)
        }));
    }

    for event in ["start", "tick", "stop"] {
        tx.send(event.to_string()).unwrap();
    }
    drop(tx);

    for handle in handles {
        let (id, seen) = handle.await.unwrap();
        println!("worker {id} saw {seen:?}");
    }
}

Output

worker 1 saw ["start", "tick", "stop"]
worker 2 saw ["start", "tick", "stop"]
worker 3 saw ["start", "tick", "stop"]

The first receiver returned by channel is discarded (_), which is allowed because the workers subscribe before any send. The while let Ok(..) loop ends on Closed. It would also end on Lagged, so a loop that must survive lag matches the error, as in Example 11.

Example 14: watch holds one current value

A watch channel stores one value. watch::channel(initial) needs an initial value, so a receiver can always read something. Receiver::borrow() returns a reference to the current value. Sender::send replaces it.

Rust

use tokio::sync::watch;

#[tokio::main]
async fn main() {
    let (tx, rx) = watch::channel("v1");
    println!("initial: {}", *rx.borrow());

    tx.send("v2").unwrap();
    println!("after send: {}", *rx.borrow());

    let other = tx.subscribe();
    println!("new receiver: {}", *other.borrow());
}

Output

initial: v1
after send: v2
new receiver: v2

borrow returns a guard (Ref) that holds a read lock on the value. Keep it short-lived and never hold it across an .await: while a Ref is alive, send can block, and the documentation warns of a possible deadlock in that case.

Example 15: changed waits for a new value, and only the latest survives

Receiver::changed().await completes when the channel holds a value this receiver has not seen. Values sent in quick succession replace each other, so a receiver that was not looking sees only the last one. The initial value counts as seen, and so does the current value of a receiver created by subscribe.

Rust

use tokio::sync::watch;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = watch::channel(0);
    println!("changed at start: {}", rx.has_changed().unwrap());

    tx.send(1).unwrap();
    tx.send(2).unwrap();
    tx.send(3).unwrap();
    println!("changed after 3 sends: {}", rx.has_changed().unwrap());

    rx.changed().await.unwrap();
    println!("value after changed: {}", *rx.borrow());
    println!("changed again: {}", rx.has_changed().unwrap());
}

Output

changed at start: false
changed after 3 sends: true
value after changed: 3
changed again: false

Three sends produced one notification, and the values 1 and 2 were never observable. Use watch for state ("the current configuration"), not for events that must all be handled.

Example 16: borrow versus borrow_and_update

borrow reads the value and leaves the "seen" mark alone, so changed still fires. borrow_and_update reads the value and marks it seen. A loop that waits with changed reads with borrow_and_update, so each value is handled once.

Rust

use tokio::sync::watch;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = watch::channel("a");

    tx.send("b").unwrap();
    println!("borrow: {}", *rx.borrow());
    println!("changed after borrow: {}", rx.has_changed().unwrap());

    println!("borrow_and_update: {}", *rx.borrow_and_update());
    println!("changed after borrow_and_update: {}", rx.has_changed().unwrap());
}

Output

borrow: b
changed after borrow: true
borrow_and_update: b
changed after borrow_and_update: false

Example 17: a receiver loop with acknowledgements

The usual receiver loop reads the current value first, then waits for changes. This is a do-while shape: the initial value is processed before the first changed().await. The sender waits for an acknowledgement on an mpsc channel after each value, so the output does not depend on scheduling and no value is skipped.

Rust

use tokio::sync::{mpsc, watch};

#[tokio::main]
async fn main() {
    let (tx, mut rx) = watch::channel(String::from("config-1"));
    let (ack_tx, mut ack_rx) = mpsc::channel::<()>(1);

    let receiver = tokio::spawn(async move {
        loop {
            let current = rx.borrow_and_update().clone();
            println!("applying {current}");
            let _ = ack_tx.send(()).await;
            if rx.changed().await.is_err() {
                println!("sender dropped, stopping");
                break;
            }
        }
    });

    ack_rx.recv().await;
    for version in 2..=3 {
        tx.send(format!("config-{version}")).unwrap();
        ack_rx.recv().await;
    }
    drop(tx);
    receiver.await.unwrap();
}

Output

applying config-1
applying config-2
applying config-3
sender dropped, stopping

The clone() ends the borrow before the .await calls, so no Ref is held across an await point.

Example 18: send_modify and send_if_modified

send_modify edits the stored value in place and always notifies receivers. send_if_modified runs a closure that returns true when it changed the value. Receivers are notified only in that case, so no-op updates cause no wake-ups. Both work even when no receiver exists.

Rust

use tokio::sync::watch;

#[derive(Debug)]
struct Settings {
    level: u8,
    name: String,
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = watch::channel(Settings { level: 1, name: "base".into() });

    tx.send_modify(|s| s.level += 1);
    println!("notified = {}", rx.has_changed().unwrap());
    println!("modify: {:?}", *rx.borrow_and_update());

    let changed = tx.send_if_modified(|s| {
        if s.name == "base" {
            false
        } else {
            s.name = "base".into();
            true
        }
    });
    println!("send_if_modified (no change): {changed}, notified = {}", rx.has_changed().unwrap());

    let changed = tx.send_if_modified(|s| {
        s.name = "custom".into();
        true
    });
    println!("send_if_modified (change): {changed}, notified = {}", rx.has_changed().unwrap());
    println!("value: {:?}", *rx.borrow_and_update());
}

Output

notified = true
modify: Settings { level: 2, name: "base" }
send_if_modified (no change): false, notified = false
send_if_modified (change): true, notified = true
value: Settings { level: 2, name: "custom" }

Example 19: closed channels and send_replace

watch::Sender::send returns Err when every receiver has been dropped, and the value is not stored. send_replace always stores the value and returns the previous one. On the receiving side, changed returns Err only when the sender is gone and the current value has been seen, so a final value sent just before the sender dropped is still delivered.

Rust

use tokio::sync::watch;

#[tokio::main]
async fn main() {
    let (tx, rx) = watch::channel(1);
    drop(rx);
    println!("send with no receivers: {:?}", tx.send(2));
    println!("value still: {}", *tx.borrow());

    let old = tx.send_replace(3);
    println!("send_replace returned {old}, value is {}", *tx.borrow());

    let mut rx = tx.subscribe();
    tx.send_replace(4);
    drop(tx);
    println!("first changed: {:?}", rx.changed().await);
    println!("value: {}", *rx.borrow_and_update());
    println!("second changed: {:?}", rx.changed().await);
}

Output

send with no receivers: Err(SendError { .. })
value still: 1
send_replace returned 1, value is 3
first changed: Ok(())
value: 4
second changed: Err(RecvError(()))

Example 20: wait_for a condition

Receiver::wait_for(predicate) checks the current value, then each new value, and returns when the predicate is true. It replaces a hand-written changed loop for "wait until the state is X".

Rust

use std::time::Duration;
use tokio::sync::watch;
use tokio::time::{self, sleep};

#[derive(Debug, PartialEq)]
enum State {
    Starting,
    Warming(u8),
    Ready,
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    time::pause();
    let (tx, mut rx) = watch::channel(State::Starting);

    tokio::spawn(async move {
        for percent in [30, 60, 90] {
            sleep(Duration::from_secs(1)).await;
            tx.send(State::Warming(percent)).unwrap();
        }
        sleep(Duration::from_secs(1)).await;
        tx.send(State::Ready).unwrap();
        sleep(Duration::from_secs(1)).await;
    });

    let state = rx.wait_for(|s| *s == State::Ready).await.unwrap();
    println!("reached {:?}", *state);
}

Output

reached Ready

wait_for returns a Ref guard, so the same rule as borrow applies: use the value and drop the guard before awaiting anything else. If the sender is dropped and no matching value ever arrived, wait_for returns Err(RecvError).

Example 21: the same event through three channels

A subscriber that joins after an event sees different things depending on the channel. This example sends one event through each channel type before the late receiver appears.

Rust

use tokio::sync::{broadcast, mpsc, watch};

#[tokio::main]
async fn main() {
    let (b_tx, _keep) = broadcast::channel(8);
    let (w_tx, _w_rx) = watch::channel("none");
    let (m_tx, mut m_rx) = mpsc::channel(8);

    b_tx.send("event-1").unwrap();
    w_tx.send("event-1").unwrap();
    m_tx.send("event-1").await.unwrap();

    let mut b_late = b_tx.subscribe();
    let w_late = w_tx.subscribe();

    println!("broadcast late receiver: {:?}", b_late.try_recv());
    println!("watch late receiver: {}", *w_late.borrow());
    println!("mpsc receiver: {:?}", m_rx.try_recv());
}

Output

broadcast late receiver: Err(Empty)
watch late receiver: event-1
mpsc receiver: Ok("event-1")

broadcast shows nothing because a receiver sees only messages sent after it subscribed. watch shows the current value. mpsc holds the message for its single receiver.

Example 22: std channels in async code

std::sync::mpsc::Receiver::recv blocks the calling thread. Inside an async task that blocks the worker thread, and every other task on that thread stalls. The example runs on the current-thread runtime, which has one thread. The sender task cannot run while the main task blocks in recv_timeout, so the wait times out.

Rust

use std::sync::mpsc;
use std::time::Duration;

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, rx) = mpsc::channel();

    tokio::spawn(async move {
        tx.send("from task").unwrap();
    });

    // Blocks the only thread. The spawned task cannot run meanwhile.
    let blocked = rx.recv_timeout(Duration::from_millis(50));
    println!("blocking recv: {blocked:?}");

    // Yielding lets the task run.
    tokio::task::yield_now().await;
    println!("after yield: {:?}", rx.try_recv());
}

Output

blocking recv: Err(Timeout)
after yield: Ok("from task")

Sending on a std channel never blocks (the unbounded mpsc::channel), so it is safe in async code. Receiving is the problem. Two fixes exist: use a Tokio channel (the receiver is .awaited), or move the blocking recv to the blocking pool with spawn_blocking (Blocking and CPU-Bound Work).

Example 23: a blocking std receiver on the blocking pool

When a library hands over a std receiver, spawn_blocking runs the blocking loop on a separate thread. The result comes back through the JoinHandle.

Rust

use std::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel::<u32>();

    let summer = tokio::task::spawn_blocking(move || {
        let mut total = 0;
        while let Ok(n) = rx.recv() {
            total += n;
        }
        total
    });

    for n in 1..=10 {
        tx.send(n).unwrap();
    }
    drop(tx);

    println!("sum = {}", summer.await.unwrap());
}

Output

sum = 55

Dropping tx ends the blocking loop, because recv returns Err once every sender is gone.

Choosing a channel

Need	Use	Reason
Many producers, one consumer, every message handled once	`mpsc` (bounded)	A queue with backpressure
One answer to one question	`oneshot`	The sender is consumed, and a dropped sender is reported
Every subscriber must see each event	`broadcast`	Per-receiver copies, lag reported
Subscribers need only the current state	`watch`	One value, changes coalesce
A value should be readable at any time without waiting	`watch` (`borrow`)	No queue, no `.await`
Late joiners need history	none of the three (keep a log, or send a snapshot on join)	`broadcast` starts at subscribe time, `watch` keeps one value
One shared counter or map, no notification needed	`Arc<Mutex<_>>` (Shared State: Mutex and RwLock)	A channel adds nothing

Property	`oneshot`	`broadcast`	`watch`	`mpsc`
Receivers	1	many	many	1
Values kept	1	last N (capacity)	1	up to capacity
Each receiver sees each value	n/a	yes, unless lagged	no, only the latest	n/a (one receiver)
`send` waits for space	no	no	no	yes (bounded)
Message type must be `Clone`	no	yes	no	no

Interview points

A oneshot::Sender::send consumes the sender and returns Err(value) if the receiver is gone. A receiver returns RecvError when the sender is dropped without sending.
The request/response pattern pairs an mpsc request channel with a oneshot reply channel inside each request.
Sender::closed() lets a producer stop work nobody is waiting for.
broadcast gives every receiver every message. It requires Clone, has a bounded ring buffer, and reports Lagged(n) to a receiver that fell behind. The receiver stays usable after Lagged.
broadcast::Sender::send returns the receiver count and fails only when there are zero receivers. A subscribe after a send does not see that message.
watch keeps the latest value. Rapid updates coalesce, so it fits state (configuration, status) and not events.
Use borrow_and_update in a changed loop. Do not hold a watch::Ref across an .await.
send_if_modified avoids waking receivers for no-op updates. send_replace and send_modify work with no receivers.
std::sync::mpsc::Receiver::recv blocks the thread. In async code, use a Tokio channel or spawn_blocking.

Summary

oneshot: one value, one sender, one receiver. The receiver is a future. Errors: RecvError (sender dropped), TryRecvError::{Empty, Closed}.
broadcast: channel(capacity), subscribe, send returns the receiver count, recv returns Ok, Lagged(n), or Closed. The buffer length is the capacity rounded up to a power of two.
watch: channel(initial), borrow, borrow_and_update, changed, has_changed, wait_for, send, send_replace, send_modify, send_if_modified.
A watch receiver treats the initial value, and the value at subscribe time, as seen.
changed returns Err when the sender is gone and the current value is already seen.
oneshot and watch send never wait. broadcast::send never waits and overwrites the oldest message when full.
Choose by shape: reply (oneshot), every event to every subscriber (broadcast), current state (watch), work queue (mpsc).

Example 1: The smallest oneshot

Example 2: send fails when the receiver is gone

Example 3: a sender cannot be used twice

Example 4: a dropped sender ends the wait

Example 5: try_recv without waiting

Example 6: request and response

Example 7: detect that the caller gave up

Example 8: a timeout around a reply

Example 9: broadcast delivers every message to every receiver

Example 10: subscribe sees only later messages

Example 11: a slow receiver lags

Example 12: closing and draining

Example 13: fan-out to several tasks

Example 14: watch holds one current value

Example 15: changed waits for a new value, and only the latest survives

Example 16: borrow versus borrow_and_update

Example 17: a receiver loop with acknowledgements

Example 18: send_modify and send_if_modified

Example 19: closed channels and send_replace

Example 20: wait_for a condition

Example 21: the same event through three channels

Example 22: std channels in async code

Example 23: a blocking std receiver on the blocking pool

Choosing a channel

Interview points

Summary