# Broadcast price ticks to every consumer with fibre spmc

**For:** Rust developers who know the basics of Tokio.

You'll build a small Tokio program where one price feed sends ticks to several consumers over a bounded `spmc` channel and every consumer sees every tick. Along the way you'll add a consumer while the feed is running, watch the slowest consumer set the pace for the feed and let consumers leave early without breaking the others. The program shuts down on its own when the feed ends or when nobody is listening any more.

Before you start: you need Rust with Cargo. fibre comes from crates.io as `fibre = "0"`.

## Set up the project

Start a new binary crate:

```sh
cargo new fibre-ticks
cd fibre-ticks
```

Add fibre, Tokio and `futures-util` to `Cargo.toml`. The program only needs Tokio's single-threaded runtime, its macros and its timers. `futures-util` provides `StreamExt`, which you'll use on a receiver in the last step.

```toml title="Cargo.toml"
[dependencies]
fibre = "0"
tokio = { version = "1", features = ["macros", "rt", "time"] }
futures-util = "0.3"
```

Every step uses the `current_thread` runtime. Everything runs on one thread, so the output comes out in the same order every time and you can compare it line for line with what's shown here.

## Send every tick to two consumers

`spmc` is fibre's broadcast channel: one sender and any number of receivers. Unlike `mpmc`, where each message goes to exactly one receiver, an `spmc` message is cloned and delivered to every receiver, so `T` has to be `Clone`. The channel is always bounded. Replace `src/main.rs` with this:

```rust title="src/main.rs"
use fibre::spmc;

#[derive(Debug, Clone)]
struct Tick {
    symbol: &'static str,
    price: f64,
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, rx) = spmc::bounded_async::<Tick>(4);

    let mut consumers = Vec::new();
    for name in ["chart", "audit"] {
        let rx = rx.clone();
        consumers.push(tokio::spawn(async move {
            while let Ok(tick) = rx.recv().await {
                println!("{name}: {} {:.2}", tick.symbol, tick.price);
            }
            println!("{name}: feed closed");
        }));
    }
    drop(rx);

    let ticks = [("ACME", 101.25), ("ACME", 101.50), ("ACME", 100.75)];
    for (symbol, price) in ticks {
        tx.send(Tick { symbol, price }).await.expect("a consumer is listening");
    }
    drop(tx);

    for consumer in consumers {
        consumer.await.unwrap();
    }
}
```

`spmc::bounded_async(4)` returns a `BoundedAsyncSender` and a `BoundedAsyncReceiver`. The sender is not `Clone`, which is what makes the channel single-producer. The receiver is `Clone` and each clone is an independent consumer with its own place in the stream. Here `chart` and `audit` each get a clone and `main` drops the original with `drop(rx)`.

> That drop matters more on `spmc` than on other channels. A receiver still counts as a consumer while it's alive, even if nobody ever calls `recv` on it. The feed can't get more than 4 ticks ahead of any consumer, so once 4 ticks had piled up behind the forgotten original, every later `send` would wait forever.

Run it:

```shell-session
$ cargo run
chart: ACME 101.25
chart: ACME 101.50
chart: ACME 100.75
chart: feed closed
audit: ACME 101.25
audit: ACME 101.50
audit: ACME 100.75
audit: feed closed
```

Both consumers got all three ticks in order. `main` sends all three before it awaits anything, so the consumers only start running once it waits on the first handle. `chart` then reads through its three ticks and `audit` reads the same three again. Reading a tick in one consumer doesn't take it away from the other.

The end of the stream works like the other fibre channels. `drop(tx)` disconnects the channel. Each receiver still drains the ticks buffered for it and only then gets `Err(RecvError::Disconnected)`, which ends its `while let` loop.

## Let the slowest consumer set the pace

A bounded broadcast channel has to keep each tick until every consumer has read it. So the feed can only run as far ahead as the slowest consumer allows. To see it, shrink the channel to 2, make `audit` take 100ms per tick and have the feed print a line each time a tick gets in. The helper `ms` rounds the elapsed time to the nearest 100ms so the numbers stay stable from run to run. Here's the whole file:

```rust title="src/main.rs"
use fibre::spmc;
use std::time::Duration;
use tokio::time::{sleep, Instant};

#[derive(Debug, Clone)]
struct Tick {
    symbol: &'static str,
    price: f64,
}

fn ms(start: Instant) -> u128 {
    (start.elapsed().as_millis() + 50) / 100 * 100
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, rx) = spmc::bounded_async::<Tick>(2);
    let start = Instant::now();

    let mut consumers = Vec::new();
    for (name, delay) in [("chart", 0), ("audit", 100)] {
        let rx = rx.clone();
        consumers.push(tokio::spawn(async move {
            while let Ok(tick) = rx.recv().await {
                sleep(Duration::from_millis(delay)).await;
                println!("{:>4}ms {name}: {} {:.2}", ms(start), tick.symbol, tick.price);
            }
            println!("{:>4}ms {name}: feed closed", ms(start));
        }));
    }
    drop(rx);

    let prices = [101.25, 101.50, 100.75, 100.50, 101.00];
    for (n, price) in prices.into_iter().enumerate() {
        tx.send(Tick { symbol: "ACME", price }).await.expect("a consumer is listening");
        println!("{:>4}ms feed: sent #{n}", ms(start));
    }
    drop(tx);

    for consumer in consumers {
        consumer.await.unwrap();
    }
}
```

```text
   0ms feed: sent #0
   0ms feed: sent #1
   0ms feed: sent #2
   0ms chart: ACME 101.25
   0ms chart: ACME 101.50
   0ms chart: ACME 100.75
 100ms audit: ACME 101.25
 100ms feed: sent #3
 100ms chart: ACME 100.50
 200ms audit: ACME 101.50
 200ms feed: sent #4
 200ms chart: ACME 101.00
 200ms chart: feed closed
 300ms audit: ACME 100.75
 400ms audit: ACME 100.50
 500ms audit: ACME 101.00
 500ms audit: feed closed
```

At time zero the feed gets three ticks in. The first two fill the buffer. Then `audit` takes `101.25` out and starts its 100ms of work, which frees one slot for tick `#2`. From there the feed's `send` waits for `audit`: `#3` gets in at 100ms when `audit` picks up its next tick and `#4` at 200ms.

`chart` does no work at all, yet it prints `100.50` at 100ms and `101.00` at 200ms. It has nothing to read in between because the feed is stuck waiting on `audit`. On an `spmc` channel every consumer runs at the speed of the slowest one, give or take the buffer. A bigger capacity absorbs short stalls. It won't help a consumer that is slower on average.

If the feed would rather drop ticks than wait, use `try_send`. It never waits. When any consumer's view of the buffer is full, it returns `TrySendError::Full` with the tick inside. Swap the `send` line and its `println!` in the feed loop for this and add `use fibre::error::TrySendError;` at the top:

```rust title="src/main.rs"
match tx.try_send(Tick { symbol: "ACME", price }) {
    Ok(()) => println!("{:>4}ms feed: sent #{n}", ms(start)),
    Err(TrySendError::Full(tick)) => println!("{:>4}ms feed: dropped #{n} at {:.2}, a consumer is behind", ms(start), tick.price),
    Err(_) => break,
}
```

```text
   0ms feed: sent #0
   0ms feed: sent #1
   0ms feed: dropped #2 at 100.75, a consumer is behind
   0ms feed: dropped #3 at 100.50, a consumer is behind
   0ms feed: dropped #4 at 101.00, a consumer is behind
   0ms chart: ACME 101.25
   0ms chart: ACME 101.50
   0ms chart: feed closed
 100ms audit: ACME 101.25
 200ms audit: ACME 101.50
 200ms audit: feed closed
```

The feed now finishes at time zero. Ticks `#2` to `#4` are gone for everyone, `chart` included, because the channel has one buffer shared by all consumers. The catch-all arm covers `TrySendError::Closed`, which means every receiver is gone. Put the blocking `send` back before moving on.

## Add a consumer while the feed runs

Consumers don't have to exist up front. Any receiver can be cloned at any time and the clone joins as a new consumer. This time the feed runs in its own task and sends a tick every 100ms, like a real feed would. `main` reads as the chart. After its second tick it clones its receiver and hands the clone to a new `recorder` task. The consumer loop moves into a `watch` function so both can share it:

```rust title="src/main.rs"
use fibre::spmc;
use std::time::Duration;
use tokio::time::sleep;

#[derive(Debug, Clone)]
struct Tick {
    symbol: &'static str,
    price: f64,
}

async fn watch(name: &'static str, rx: spmc::BoundedAsyncReceiver<Tick>) {
    while let Ok(tick) = rx.recv().await {
        println!("{name}: {} {:.2}", tick.symbol, tick.price);
    }
    println!("{name}: feed closed");
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, rx) = spmc::bounded_async::<Tick>(4);

    let feed = tokio::spawn(async move {
        let prices = [101.25, 101.50, 100.75, 100.50, 101.00];
        for price in prices {
            sleep(Duration::from_millis(100)).await;
            tx.send(Tick { symbol: "ACME", price }).await.expect("a consumer is listening");
        }
    });

    let mut recorder = None;
    let mut seen = 0;
    while let Ok(tick) = rx.recv().await {
        println!("chart: {} {:.2}", tick.symbol, tick.price);
        seen += 1;
        if seen == 2 {
            recorder = Some(tokio::spawn(watch("recorder", rx.clone())));
        }
    }
    println!("chart: feed closed");

    feed.await.unwrap();
    if let Some(recorder) = recorder {
        recorder.await.unwrap();
    }
}
```

```text
chart: ACME 101.25
chart: ACME 101.50
recorder: ACME 100.75
chart: ACME 100.75
recorder: ACME 100.50
chart: ACME 100.50
recorder: ACME 101.00
recorder: feed closed
chart: ACME 101.00
chart: feed closed
```

`recorder` starts with the third tick. A clone starts at the same point in the stream as the receiver it was cloned from. `rx` had already read `101.25` and `101.50` when it was cloned, so those two are not repeated for `recorder`. Everything sent after that reaches both.

From then on each tick shows up in both consumers. Both are waiting when a tick arrives, so which of the two prints it first depends on the order they are woken in, which fibre doesn't promise. What stays fixed is the order of ticks within each consumer and the point where `recorder` joins. When the feed task finishes, its sender is dropped and both loops end after the last tick.

## Let consumers leave early

A consumer can stop whenever it likes. Dropping its receiver removes it from the channel. The feed stops waiting on it and the other consumers carry on. When the last receiver is gone, `send` returns `Err(SendError::Closed)` and the feed knows there is no point in continuing.

The finished program below has two consumers that each leave early. `alerts` waits for the first price under 101.00, reports it and exits. `BoundedAsyncReceiver` implements `futures::Stream`, so it uses `StreamExt::filter` and `next` instead of a `recv` loop. `chart` reads four ticks and then drops its receiver. Replace all of `src/main.rs` with it:

```rust title="src/main.rs"
use fibre::spmc;
use futures_util::StreamExt;
use std::future::ready;
use std::time::Duration;
use tokio::time::sleep;

#[derive(Debug, Clone)]
struct Tick {
    symbol: &'static str,
    price: f64,
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, rx) = spmc::bounded_async::<Tick>(4);

    let feed = tokio::spawn(async move {
        let prices = [101.25, 101.50, 100.75, 100.50, 101.00, 101.75];
        for (n, price) in prices.into_iter().enumerate() {
            sleep(Duration::from_millis(100)).await;
            if tx.send(Tick { symbol: "ACME", price }).await.is_err() {
                println!("feed: no consumers left, stopping at #{n}");
                return;
            }
        }
        println!("feed: all ticks sent");
    });

    let alerts_rx = rx.clone();
    let alerts = tokio::spawn(async move {
        let mut low = alerts_rx.filter(|tick| ready(tick.price < 101.00));
        if let Some(tick) = low.next().await {
            println!("alerts: {} fell to {:.2}, alert sent", tick.symbol, tick.price);
        }
    });

    let mut seen = 0;
    while let Ok(tick) = rx.recv().await {
        println!("chart: {} {:.2}", tick.symbol, tick.price);
        seen += 1;
        if seen == 4 {
            break;
        }
    }
    drop(rx);
    println!("chart: closed after {seen} ticks");

    alerts.await.unwrap();
    feed.await.unwrap();
}
```

```text
chart: ACME 101.25
chart: ACME 101.50
alerts: ACME fell to 100.75, alert sent
chart: ACME 100.75
chart: ACME 100.50
chart: closed after 4 ticks
feed: no consumers left, stopping at #4
```

As in the last step, which consumer prints first for the same tick can vary. `alerts` sees `101.25` and `101.50`, skips them, returns on `100.75` and drops `low` along with the receiver inside it. The feed keeps going without it and `chart` still gets its fourth tick. After that `chart` breaks out of its loop and `drop(rx)` removes the last receiver. The feed's next `send` at 500ms fails, so it stops at `#4` instead of sending into a channel nobody reads.

## Shut down in order

An `spmc` channel has the same two shutdown paths as the other fibre channels. The difference is how many consumers each one reaches:

```text
feed drops tx        -> every receiver drains its buffered ticks, then recv() returns Err
a consumer drops rx  -> that consumer stops at once, the feed and other consumers carry on
last rx dropped      -> the feed's next send() returns Err(SendError::Closed)
```

Closing from the sending side is the clean stop. Each consumer finishes every tick that was sent before the close, however far behind it was. Dropping a receiver (or calling `close()` on it) doesn't drain anything. That consumer just stops reading, which is fine for a consumer that has got what it came for, like `alerts`.

The sender has an explicit `close()` too. It takes `&mut self`, so the binding has to be mutable. In the first step's program, change the `let` line and replace `drop(tx)` with the `close()` call:

```rust title="src/main.rs"
let (mut tx, rx) = spmc::bounded_async::<Tick>(4);

tx.close().expect("the feed closes once");
```

The output is the same as before. Consumers see the same thing as with a dropped sender: they drain what's buffered and then get `Err(RecvError::Disconnected)`. Keep the rule from the first step in mind as well. Every receiver you keep around, even one you only hold to clone later, is a consumer the feed has to wait for.

## Where to go next

The channels guide at [https://github.com/excsn/fibre/blob/main/channels/README.GUIDE.md](https://github.com/excsn/fibre/blob/main/channels/README.GUIDE.md) covers the rest of fibre. For broadcast it also has `spmc::topic`, a publish-subscribe variant where consumers subscribe to named topics and a slow consumer loses messages instead of holding up the sender. It also covers the batch API (`send_batch` and `recv_batch` work on `spmc` too) and mixing sync threads with async tasks on one channel using `to_sync()` and `to_async()`, for example to hand a receiver clone to a blocking thread that writes ticks to disk.

The full API, including every error type, is on docs.rs at [https://docs.rs/fibre](https://docs.rs/fibre).