use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
use axum::extract::State;
use axum::response::Response;
use axum::routing::get;
use axum::Router;
use crate::event_hub::{AppEvent, EventHub};
use crate::subscriber_task::{run_subscriber, EventSink};
pub fn router(hub: EventHub) -> Router {
Router::new().route("/events", get(ws_handler)).with_state(hub)
}
async fn ws_handler(ws: WebSocketUpgrade, State(hub): State<EventHub>) -> Response {
let rx = hub.subscribe();
ws.on_upgrade(move |socket| async move {
run_subscriber(rx, WsSink(socket)).await;
})
}
struct WsSink(WebSocket);
impl EventSink for WsSink {
async fn deliver(&mut self, event: AppEvent) -> Result<(), ()> {
let payload = serde_json::to_string(&event).map_err(|_| ())?;
self.0.send(Message::Text(payload)).await.map_err(|_| ())
}
}
use tokio::sync::broadcast;
#[derive(Clone, Debug)]
pub enum AppEvent {
UserJoined { id: u64, name: String },
PriceTick { symbol: String, price: f64 },
JobFinished { job_id: String },
}
#[derive(Clone)]
pub struct EventHub {
sender: broadcast::Sender<AppEvent>,
}
impl EventHub {
pub fn new(capacity: usize) -> Self {
let (sender, _rx) = broadcast::channel(capacity);
EventHub { sender }
}
pub fn subscribe(&self) -> broadcast::Receiver<AppEvent> {
self.sender.subscribe()
}
pub fn publish(&self, event: AppEvent) {
// A send with no live receivers is normal for a broadcast hub.
let _ = self.sender.send(event);
}
pub fn subscriber_count(&self) -> usize {
self.sender.receiver_count()
}
}
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::broadcast::Receiver;
use crate::event_hub::AppEvent;
pub async fn run_subscriber<S>(mut rx: Receiver<AppEvent>, mut sink: S)
where
S: EventSink,
{
loop {
match rx.recv().await {
Ok(event) => {
if sink.deliver(event).await.is_err() {
break;
}
}
Err(RecvError::Lagged(skipped)) => {
tracing::warn!(skipped, "subscriber lagged, dropping messages");
continue;
}
Err(RecvError::Closed) => break,
}
}
}
pub trait EventSink {
async fn deliver(&mut self, event: AppEvent) -> Result<(), ()>;
}
This snippet shows a small pub/sub hub built on tokio::sync::broadcast, the standard tool for delivering the same message to many independent consumers at once. Unlike an mpsc channel, which fans multiple producers into a single receiver, a broadcast channel fans one send into every live Receiver. That makes it a natural fit for pushing domain events (a user joined, a price ticked, a job finished) out to any number of connected clients without the sender knowing who is listening.
In event_hub.rs, EventHub wraps only the broadcast::Sender. New subscribers are minted on demand via subscribe(), which calls sender.subscribe() to hand back a fresh Receiver. The Sender is cheap to clone and can be shared across tasks, so the hub itself is Clone and behaves like a handle. publish intentionally ignores the SendError returned when there are zero receivers — with a broadcast channel a send with no listeners is a normal, non-fatal condition, not something callers should crash on.
The crucial detail is the bounded ring buffer sized by broadcast::channel(capacity). Each receiver tracks its own read cursor; a slow consumer that falls more than capacity messages behind will have the oldest messages overwritten and will observe RecvError::Lagged(n) on its next recv(). This is deliberate backpressure-by-dropping: one stuck client can never stall the publisher or the other subscribers. The trade-off is that lagging clients lose messages, so Lagged should be handled explicitly rather than treated as a hard error.
In subscriber_task.rs, run_subscriber demonstrates the correct receive loop: Closed breaks the loop, Lagged(skipped) is logged and the loop continues from the newest available message, and a normal event is forwarded to the socket. Cloning the AppEvent is fine because broadcast requires T: Clone to deliver a copy to every receiver.
Finally, routes.rs wires the hub into an axum WebSocket handler. The EventHub lives in shared state, each upgraded connection calls subscribe() to get its own receiver, and a per-connection task drives run_subscriber. Elsewhere in the app, any code holding a cloned EventHub can call publish and reach every open socket. This pattern suits live dashboards and notification feeds where dropping stale updates under load is acceptable; when no message may ever be lost, a durable queue is the better choice.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
class Comment < ApplicationRecord
belongs_to :article
belongs_to :author, class_name: "User"
validates :body, presence: true, length: { maximum: 2_000 }Live comments with model broadcasts + turbo_stream_from
Share this code
Here's the card — post it anywhere.
