# Build a topic event bus with fibre

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

You'll build a small in-process event bus for a shop. One publisher sends events under the topic keys `orders`, `payments` and `shipping`. A handful of subscribers each receive only the topics they asked for. Along the way you'll change subscriptions while the bus is running, watch what happens to a subscriber that can't keep up and add a plain thread as a subscriber next to the async ones.

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-bus
cd fibre-bus
```

Add fibre and Tokio to `Cargo.toml`. The bus only needs Tokio's single-threaded runtime, its macros and its timers.

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

Every async step uses the `current_thread` runtime. Tasks only switch at an `.await`, so the output comes out in the same order every time and you can compare it line for line with what's shown here.

## Publish to one subscriber

The channel for this is `fibre::spmc::topic`. It has one sender and any number of receivers. The sender attaches a topic key to every message and each receiver says which keys it wants with `subscribe`. Replace `src/main.rs` with this:

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

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (bus, billing) = topic::channel_async::<&'static str, String>(16);
    billing.subscribe("orders");

    let handle = tokio::spawn(async move {
        while let Ok((topic, event)) = billing.recv().await {
            println!("billing: [{topic}] {event}");
        }
        println!("billing: bus closed, exiting");
    });

    bus.send("orders", "order 1 placed".to_string()).unwrap();
    bus.send("shipping", "order 1 packed".to_string()).unwrap();
    bus.send("orders", "order 2 placed".to_string()).unwrap();
    drop(bus);

    handle.await.unwrap();
}
```

`topic::channel_async::<K, T>(16)` returns an `AsyncTopicSender` and an `AsyncTopicReceiver`. `K` is the topic key type and `T` is the message type. Both must be `Clone`, because one message can go to several subscribers and each one gets its own copy. The `16` is the mailbox capacity. Every receiver has a private mailbox of that size, which comes up again later.

`send` takes the topic and the message. It is not `async` and it never waits. It puts a copy of the message in the mailbox of every receiver subscribed to that topic and returns. Nobody is subscribed to `shipping` yet, so `order 1 packed` goes nowhere.

`recv` resolves with a `(topic, message)` pair. Once the sender has been dropped and this receiver's mailbox is empty, it resolves with `Err(RecvError::Disconnected)` instead, which ends the `while let` loop. Run it:

```shell-session
$ cargo run
billing: [orders] order 1 placed
billing: [orders] order 2 placed
billing: bus closed, exiting
```

The spawned task doesn't run until `main` reaches `handle.await`. By then all three events have been sent and the sender is gone, so billing drains its two orders and then sees the bus close.

## Give each subscriber its own topics

Receivers are `Clone`. A clone is a separate subscriber with its own mailbox and its own set of topics, so each part of the shop gets one. Billing cares about orders and payments. The warehouse cares about orders and shipping. Replace `src/main.rs` with this version, which moves the receive loop into a small helper:

```rust title="src/main.rs"
use fibre::spmc::topic::{self, AsyncTopicReceiver};
use tokio::task::JoinHandle;

fn subscriber(name: &'static str, rx: AsyncTopicReceiver<&'static str, String>) -> JoinHandle<()> {
    tokio::spawn(async move {
        while let Ok((topic, event)) = rx.recv().await {
            println!("{name}: [{topic}] {event}");
        }
        println!("{name}: bus closed, exiting");
    })
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (bus, billing) = topic::channel_async::<&'static str, String>(16);
    let warehouse = billing.clone();

    billing.subscribe("orders");
    billing.subscribe("payments");
    warehouse.subscribe("orders");
    warehouse.subscribe("shipping");

    let handles = [subscriber("billing", billing), subscriber("warehouse", warehouse)];

    let events = [
        ("orders", "order 1 placed"),
        ("payments", "order 1 paid"),
        ("shipping", "order 1 packed"),
        ("orders", "order 2 placed"),
        ("payments", "order 2 paid"),
        ("shipping", "order 1 delivered"),
    ];
    for (topic, event) in events {
        bus.send(topic, event.to_string()).unwrap();
    }
    drop(bus);

    for handle in handles {
        handle.await.unwrap();
    }
}
```

The clone is made before anything is subscribed. Then each handle subscribes to its own pair of topics. `orders` goes to both.

```text
billing: [orders] order 1 placed
billing: [payments] order 1 paid
billing: [orders] order 2 placed
billing: [payments] order 2 paid
billing: bus closed, exiting
warehouse: [orders] order 1 placed
warehouse: [shipping] order 1 packed
warehouse: [orders] order 2 placed
warehouse: [shipping] order 1 delivered
warehouse: bus closed, exiting
```

Both subscribers got `order 1 placed` and `order 2 placed`, each as its own `String`. Payments went only to billing and shipping only to the warehouse. Billing prints all of its events before the warehouse starts because neither task runs until `main` awaits the first handle. Billing was spawned first, so it runs first.

The sender is not `Clone`. A topic channel has exactly one publisher. If several parts of your program need to publish, they send to the task that owns the sender, for example over an `mpsc` channel.

## Change subscriptions at runtime

A subscriber can call `subscribe` and `unsubscribe` at any time, including from inside its own receive loop. Here a tracker follows the orders topic. When order 2 is placed it starts following shipping as well. Once order 2 is delivered it stops. The publisher now sleeps 10ms after each event, standing in for events that arrive over time, which gives the tracker a chance to react before the next one is sent:

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

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (bus, tracker) = topic::channel_async::<&'static str, String>(16);
    tracker.subscribe("orders");

    let handle = tokio::spawn(async move {
        while let Ok((topic, event)) = tracker.recv().await {
            println!("tracker: [{topic}] {event}");
            if event == "order 2 placed" {
                tracker.subscribe("shipping");
                println!("tracker: following shipping");
            }
            if event == "order 2 delivered" {
                tracker.unsubscribe(&"shipping");
                println!("tracker: stopped following shipping");
            }
        }
        println!("tracker: bus closed, exiting");
    });

    let events = [
        ("orders", "order 1 placed"),
        ("shipping", "order 1 packed"),
        ("orders", "order 2 placed"),
        ("shipping", "order 1 delivered"),
        ("shipping", "order 2 packed"),
        ("shipping", "order 2 delivered"),
        ("shipping", "order 3 packed"),
        ("orders", "order 3 placed"),
    ];
    for (topic, event) in events {
        bus.send(topic, event.to_string()).unwrap();
        sleep(Duration::from_millis(10)).await;
    }
    drop(bus);

    handle.await.unwrap();
}
```

`unsubscribe` takes the key by reference. The key type here is `&'static str`, so the argument is `&"shipping"`.

```text
tracker: [orders] order 1 placed
tracker: [orders] order 2 placed
tracker: following shipping
tracker: [shipping] order 1 delivered
tracker: [shipping] order 2 packed
tracker: [shipping] order 2 delivered
tracker: stopped following shipping
tracker: [orders] order 3 placed
tracker: bus closed, exiting
```

A subscription only covers messages sent after it. `order 1 packed` was published before the tracker subscribed to shipping, so it never arrived. From `order 1 delivered` on, the tracker saw every shipping event, including ones for other orders, because subscriptions are by topic. After the unsubscribe, `order 3 packed` was skipped while `order 3 placed` still came through on `orders`.

## See what happens to a slow subscriber

The sender never waits. It doesn't wait for a slow subscriber either. When a subscriber's mailbox is full, new messages for that subscriber are dropped and `send` still returns `Ok`. Other subscribers are not affected.

To see it, shrink the mailboxes to 2 and add an analytics subscriber that takes 350ms per event. The warehouse still handles events instantly. The publisher sends six orders, one every 100ms:

```rust title="src/main.rs"
use fibre::spmc::topic::{self, AsyncTopicReceiver};
use std::time::Duration;
use tokio::task::JoinHandle;
use tokio::time::sleep;

fn subscriber(name: &'static str, rx: AsyncTopicReceiver<&'static str, String>, work: Duration) -> JoinHandle<()> {
    tokio::spawn(async move {
        while let Ok((_, event)) = rx.recv().await {
            println!("{name}: {event}");
            sleep(work).await;
        }
        println!("{name}: bus closed, exiting");
    })
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (bus, warehouse) = topic::channel_async::<&'static str, String>(2);
    let analytics = warehouse.clone();
    warehouse.subscribe("orders");
    analytics.subscribe("orders");

    let handles = [
        subscriber("warehouse", warehouse, Duration::ZERO),
        subscriber("analytics", analytics, Duration::from_millis(350)),
    ];

    for id in 1..=6 {
        bus.send("orders", format!("order {id} placed")).unwrap();
        sleep(Duration::from_millis(100)).await;
    }
    drop(bus);

    for handle in handles {
        handle.await.unwrap();
    }
}
```

```text
warehouse: order 1 placed
analytics: order 1 placed
warehouse: order 2 placed
warehouse: order 3 placed
warehouse: order 4 placed
analytics: order 2 placed
warehouse: order 5 placed
warehouse: order 6 placed
warehouse: bus closed, exiting
analytics: order 3 placed
analytics: order 5 placed
analytics: bus closed, exiting
```

The warehouse got all six orders. Analytics took `order 1` at once and was busy until 350ms. Orders 2 and 3 filled its two-slot mailbox at 100ms and 200ms, so `order 4` was dropped at 300ms. At 350ms analytics took order 2, which freed one slot for `order 5` at 400ms. The mailbox was full again when `order 6` came at 500ms. After the sender was dropped, analytics still worked through what was already in its mailbox before it saw the bus close.

This is the trade the topic channel makes. A slow subscriber loses messages instead of slowing down the publisher or anyone else. That fits events where the latest state matters more than every step, such as metrics, progress updates or cache invalidation. If a consumer must see every message, give it a larger mailbox or send its messages on a bounded `mpsc` or `mpmc` channel, which make the sender wait.

## Add a subscriber on a plain thread

Some consumers are blocking code, like an audit log that writes to disk. `to_sync()` turns an async receiver into a `TopicReceiver` whose `recv` blocks the calling thread, so it can live on a `std::thread` next to the async subscribers. This is the finished bus. Replace all of `src/main.rs` with it:

```rust title="src/main.rs" {2-3,19,25-27,29-35,54-55}
use fibre::spmc::topic::{self, AsyncTopicReceiver};
use std::collections::BTreeMap;
use std::thread;
use tokio::task::JoinHandle;

fn subscriber(name: &'static str, rx: AsyncTopicReceiver<&'static str, String>) -> JoinHandle<()> {
    tokio::spawn(async move {
        while let Ok((topic, event)) = rx.recv().await {
            println!("{name}: [{topic}] {event}");
        }
        println!("{name}: bus closed, exiting");
    })
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (bus, billing) = topic::channel_async::<&'static str, String>(16);
    let warehouse = billing.clone();
    let audit = billing.clone().to_sync();

    billing.subscribe("orders");
    billing.subscribe("payments");
    warehouse.subscribe("orders");
    warehouse.subscribe("shipping");
    for topic in ["orders", "payments", "shipping"] {
        audit.subscribe(topic);
    }

    let audit = thread::spawn(move || {
        let mut counts = BTreeMap::new();
        while let Ok((topic, _)) = audit.recv() {
            *counts.entry(topic).or_insert(0) += 1;
        }
        counts
    });
    let handles = [subscriber("billing", billing), subscriber("warehouse", warehouse)];

    let events = [
        ("orders", "order 1 placed"),
        ("payments", "order 1 paid"),
        ("shipping", "order 1 packed"),
        ("orders", "order 2 placed"),
        ("payments", "order 2 paid"),
        ("shipping", "order 1 delivered"),
    ];
    for (topic, event) in events {
        bus.send(topic, event.to_string()).unwrap();
    }
    drop(bus);

    for handle in handles {
        handle.await.unwrap();
    }
    let counts = audit.join().unwrap();
    println!("audit: {counts:?}");
}
```

The audit receiver is a clone like the others, converted with `to_sync()` and subscribed to all three topics. Its thread counts events per topic and returns the counts when the bus closes. `main` joins it last and prints the result, so its output always comes at the end.

```text
billing: [orders] order 1 placed
billing: [payments] order 1 paid
billing: [orders] order 2 placed
billing: [payments] order 2 paid
billing: bus closed, exiting
warehouse: [orders] order 1 placed
warehouse: [shipping] order 1 packed
warehouse: [orders] order 2 placed
warehouse: [shipping] order 1 delivered
warehouse: bus closed, exiting
audit: {"orders": 2, "payments": 2, "shipping": 2}
```

The thread and the async tasks use the same channel with no adapter in between. The conversion works the other way too: `to_async()` turns a sync receiver or sender into its async form.

## Shut down the bus

Shutdown starts at the sender. Dropping it closes the bus for every subscriber. Each one keeps receiving until its own mailbox is empty and then gets `Err(RecvError::Disconnected)`, from the blocking `recv` on the audit thread as well as the async ones. You can see that in every run above: the "bus closed" line always comes after the last buffered event. The sender also has a `close()` method if you want to close the bus without dropping the handle.

The other direction goes through `send`. A subscriber that is done with the bus can simply be dropped. Once every receiver is gone, `send` returns `Err(SendError::Closed)`. Put this in `src/bin/closed.rs`. It uses the sync constructor `topic::channel`, which needs no runtime:

```rust title="src/bin/closed.rs"
use fibre::spmc::topic;

fn main() {
    let (bus, billing) = topic::channel::<&'static str, String>(16);
    billing.subscribe("orders");
    drop(billing);

    let result = bus.send("orders", "order 3 placed".to_string());
    println!("send after every subscriber left: {result:?}");
}
```

```shell-session
$ cargo run --bin closed
send after every subscriber left: Err(Closed)
```

So a publisher can tell when nobody is listening at all. It can't tell whether one particular subscriber dropped a message because its mailbox was full, since `send` returns `Ok` in that case. Every `unwrap()` on `send` in this tutorial is safe because the subscribers outlive the publisher. In a long-running service, stop publishing on `SendError::Closed` instead.

## 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: plain `spmc` for broadcasting every message to every consumer, `mpsc` and `mpmc` for queues that apply backpressure, `oneshot` for request and reply and the batch API. `AsyncTopicReceiver` also implements `futures::Stream`, so you can read a subscription with `StreamExt` combinators, which the guide shows for the other channel types.

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