Rust · topics · bounded queues
Riverbed is a message broker in Rust that delivers to every subscriber of a topic, matches wildcards like logs.*, and bounds every subscriber's queue so a slow one can never grow memory without limit. This is the real broker logic, running in your browser.
bounded queues, depth 8 each · drop-oldest policy · toggle "slow" on a subscriber, stop draining it, then publish to watch it fill and drop
Type a topic (like logs.error) or click one of the topic chips, then press Publish. The message is delivered to every subscriber whose pattern matches, and each subscriber's queue and delivered count update live.
Use + subscriber to add one with its own pattern (wildcards like logs.* match a single segment). Toggle a subscriber to slow and stop draining it, then keep publishing to watch its bounded queue fill to depth 8 and start dropping the oldest messages, while the other subscribers keep up.
The same rules as src/broker.rs and src/topic.rs, ported line for line into the page above.
A publish names a topic, a plain dot-separated string like orders.created.
A subscription pattern's * matches exactly one segment in that position, so logs.* matches logs.error but not logs.error.detail.
Every subscriber whose pattern matches gets the message enqueued, independent of every other subscriber.
Each subscriber has a max depth. A full queue never grows, it drops the oldest message or rejects the new one.
Delivered and dropped counts are tracked per subscriber and per topic, so backpressure is visible, not silent.
Removing a subscriber stops delivery to it immediately, its queue is discarded.
Why bounded queues matter
A single stalled subscriber can accumulate messages forever, and the broker's memory grows without limit until the process falls over.
Every subscriber has a fixed max depth. Past that, the broker drops the oldest queued message or rejects the new one, your choice per subscriber, and the queue simply stops growing.
Dropping is not free, a slow subscriber does lose messages. Riverbed counts every delivery and every drop, per subscriber and per topic, so the cost is a number you can see, not a surprise.
The scripted demo runs the same scenario as above: a slow subscriber with depth 4 that never drains, twenty messages published, and the delivered and dropped counts printed so the drops are a number, not a guess.
$ cargo run -- demo
riverbed demo
subscribers: fast=1 (logs.error, depth 100, reject-new)
wildcard=2 (logs.*, depth 100, reject-new)
slow=3 (logs.error, depth 4, drop-oldest) <- never drains, will overflow
publishing 20 messages to logs.error ...
publishing 5 messages to logs.info (wildcard-only, fast/slow do not match) ...
results
-------
fast delivered=20 dropped=0 queue_len=0
wildcard delivered=25 dropped=0 queue_len=0
slow delivered=20 dropped=16 queue_len=4
fast drained 20 messages, wildcard drained 25 messages before this point
topic logs.error: delivered=60 dropped=16
topic logs.info: delivered=5 dropped=0
unsubscribing slow ...
slow no longer exists, message delivered only to fast and wildcard
riverbed demo runs the scripted scenario above and prints the per-subscriber and per-topic delivered and dropped counts.
Construct a Broker with a max message size, then subscribe, publish, receive, and unsubscribe. Oversized messages and unknown ids return a typed BrokerError, never a panic.
cargo test covers exact and wildcard delivery, that a bounded queue drops or rejects at capacity, that the counters are correct, and that an unsubscribed subscriber receives nothing.
// as a library let mut broker = Broker::new(4096); // max message size in bytes let id = broker.subscribe("logs.*", 100, OverflowPolicy::DropOldest)?; broker.publish("logs.error", b"disk full".to_vec())?; while let Some(msg) = broker.receive(id)? { println!("{:?}", msg); } broker.unsubscribe(id)?;