Broadcast across instances

Publish a message once and stream it to every subscriber on every instance, with resume after a reconnect.

With one server, sse::channel is enough. With several instances behind a load balancer, the browser might reconnect to a different one, and a message published on one instance must reach clients connected to the others. A broadcast::Hub does both. Publishing writes to a shared log; every instance streams from it, and a browser that reconnects with Last-Event-ID gets what it missed from whichever instance it reaches.

src/main.rs

use axeon::broadcast::{Hub, MemoryBackend, Message};
use axeon::{Json, Server, ServerError};
use axeon::json;

fn main() {
    // One process: MemoryBackend. Several: RedisBroadcast or PgBroadcast.
    let hub = Hub::new(MemoryBackend::new());

    let mut app = Server::new();
    app.state(hub);

    // Clients subscribe to a topic. `last_event_id` resumes after a reconnect.
    app.get("/runs/:run/events", |req| async move {
        let run: u64 = req.param("run")?;
        let hub = req.state::<Hub>().unwrap();
        hub.sse(&format!("run:{}", run), req.last_event_id()).await
    });

    // Anything can publish: a handler, a background job, another instance.
    app.post("/runs/:run/steps/:step", |req| async move {
        let run: u64 = req.param("run")?;
        let step: u32 = req.param("step")?;
        let hub = req.state::<Hub>().unwrap();
        let message = Message::new(format!(r#"{{"step":{}}}"#, step)).event("step");
        hub.publish(&format!("run:{}", run), message)
            .await
            .map_err(|e| ServerError::InternalError(e.to_string()))?;
        Ok::<_, ServerError>(Json(json!({ "published": step })))
    });

    app.listen("127.0.0.1:3000").expect("server failed to start");
}

Terminal

curl -N http://127.0.0.1:3000/runs/42/events &
curl -X POST http://127.0.0.1:3000/runs/42/steps/3

Notes#

If the client's last message has already been trimmed from the log, the stream starts with a reset event so the client can reload its state instead of showing a gap.
Slow subscribers catch up from the log instead of dropping messages, and each instance uses one wake-up connection no matter how many clients it serves.
MemoryBackend is for one process and tests. For several instances use axeon_redis::RedisBroadcast (see Redis) or axeon_orm::broadcast::PgBroadcast, which publishes inside your database transaction so a message goes out only if the commit succeeds.
For a single instance with no resume, plain server-sent events are simpler.