An In Process Event Bus in Rust With Broadcast Channels

As an application grows, one action starts to trigger many reactions: an order is placed, so the confirmation email is queued, the cache is cleared, the audit log is written and the metrics counter increments. Calling all of them from the order code couples it to every consumer. An event bus inverts that: the order code publishes OrderPlaced, and interested modules subscribe. Within one process this takes about forty lines. Versions: Rust 1.90, Tokio 1.47.

The bus

use tokio::sync::broadcast;

#[derive(Debug, Clone)]
pub enum Event {
    OrderPlaced { order_id: u64, user_id: u64, total_cents: i64 },
    OrderCancelled { order_id: u64 },
    UserRegistered { user_id: u64 },
}

#[derive(Clone)]
pub struct EventBus {
    tx: broadcast::Sender<Event>,
}

impl EventBus {
    pub fn new(capacity: usize) -> Self {
        let (tx, _) = broadcast::channel(capacity);
        Self { tx }
    }

    pub fn publish(&self, event: Event) {
        // Err only means nobody is subscribed
        let _ = self.tx.send(event);
    }

    pub fn subscribe(&self) -> broadcast::Receiver<Event> {
        self.tx.subscribe()
    }
}

EventBus is cheap to clone (the sender is reference counted), so each module holds its own handle, typically passed in at construction (dependency injection). Event derives Clone because each subscriber gets its own copy.

Subscribers

A subscriber is a task with a receive loop:

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

pub fn spawn_audit_logger(bus: &EventBus) -> tokio::task::JoinHandle<()> {
    let mut rx = bus.subscribe();
    tokio::spawn(async move {
        loop {
            match rx.recv().await {
                Ok(Event::OrderPlaced { order_id, total_cents, .. }) => {
                    tracing::info!(order_id, total_cents, "order placed");
                }
                Ok(_) => {}
                Err(RecvError::Lagged(missed)) => {
                    tracing::warn!(missed, "audit logger fell behind and missed events");
                }
                Err(RecvError::Closed) => break,
            }
        }
    })
}

pub fn spawn_cache_invalidator(bus: &EventBus, cache: moka::future::Cache<u64, String>) {
    let mut rx = bus.subscribe();
    tokio::spawn(async move {
        while let Ok(event) = rx.recv().await {
            if let Event::OrderCancelled { order_id } = event {
                cache.invalidate(&order_id).await;
            }
        }
    });
}

Each subscriber filters for the variants it cares about and ignores the rest. Adding a new reaction means adding a subscriber, and the producer does not change. That is the decoupling.

Publishing

pub struct OrderService {
    bus: EventBus,
}

impl OrderService {
    pub async fn place(&self, user_id: u64, total_cents: i64) -> u64 {
        let order_id = 1001; // persisted in a real service
        self.bus.publish(Event::OrderPlaced { order_id, user_id, total_cents });
        order_id
    }
}

publish is synchronous and never waits for subscribers, so a slow consumer cannot slow the order path. That is both the strength and the hazard.

Lag: the central trade-off

The channel holds capacity events. If a subscriber falls behind by more than that, the oldest events are discarded for it and its next recv returns Lagged(n). The bus does not block the producer to protect slow consumers. Consequences:

• Choose capacity for bursts: hundreds or thousands for busy buses.
• Subscribers must be fast: do work in the loop only if it is quick, or hand off to a bounded queue of their own and handle overload explicitly (backpressure design).
• Lag is a signal. Log it, count it, alert on it.
• A subscriber that must never miss an event cannot rely on this bus.

For that case use an mpsc channel per critical subscriber (one sender per consumer, held by the bus), which applies backpressure, or a durable queue (background jobs, Tokio channels compared).

Ordering and delivery

Within one subscriber, events arrive in publication order. Across subscribers there is no ordering. Delivery is at most once per subscriber: no acknowledgements, no retries. Subscribers added later see only later events; a subscriber that starts after UserRegistered was published never sees it. If a handler panics, only its task dies, so supervise: restart the task or log loudly.

Typed topics instead of one enum

One enum is easy and keeps a single channel. If different events have different audiences and rates, run separate buses (OrderBus, UserBus) so a flood of one type does not push another's events out of a shared buffer. Alternatively use a HashMap<TypeId, broadcast::Sender<...>> for a type-keyed bus; the added complexity is rarely worth it.

Testing

#[tokio::test]
async fn subscribers_receive_published_events() {
    let bus = EventBus::new(16);
    let mut a = bus.subscribe();
    let mut b = bus.subscribe();

    bus.publish(Event::UserRegistered { user_id: 7 });

    assert!(matches!(a.recv().await.unwrap(), Event::UserRegistered { user_id: 7 }));
    assert!(matches!(b.recv().await.unwrap(), Event::UserRegistered { user_id: 7 }));
}

#[tokio::test]
async fn slow_subscriber_sees_lag_not_a_hang() {
    let bus = EventBus::new(2);
    let mut rx = bus.subscribe();
    for i in 0..5 {
        bus.publish(Event::OrderCancelled { order_id: i });
    }
    assert!(matches!(rx.recv().await, Err(broadcast::error::RecvError::Lagged(3))));
    assert!(matches!(rx.recv().await.unwrap(), Event::OrderCancelled { order_id: 3 }));
}

The second test pins the lag behavior, which is the contract your subscribers must handle.

When not to use an event bus

If one specific thing must happen after another, call it directly: events hide control flow and make "what runs when X happens?" harder to answer. Use the bus for optional side effects and notifications, not for the core of a transaction. When the work must be reliable, record the event in the database in the same transaction and process it asynchronously (outbox pattern). Actors are an alternative structure for owned state (actor model). For cross-process events see Kafka and NATS compared, and for the fan-out design in pipelines fan-out and fan-in.