Build a topic event bus with fibre

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.

fibre = "0"

Set up the project#

Start a new binary crate:

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.

[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:

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:

$ 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:

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.

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:

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".

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:

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();
    }
}

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:

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.

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:

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:?}");
}

$ 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 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.