Server-Sent Events

Server-Sent Events (SSE) is the minimum one-way push channel from server to browser: the browser opens EventSource(url), the server keeps a text/event-stream response open, and pushes framed events as they happen. No WebSocket handshake, no permessage-deflate, no framing libraries - just data:, event:, id:, retry: lines terminated by a blank line, per the WHATWG EventSource specification.

Suprnova's SSE primitive plugs into the streaming-body path: build a Stream<Item = SseEvent>, hand it to HttpResponse::sse(...), and the framework owns connection management, framing, headers, and panic isolation. The connection stays open until the producing stream ends or the client disconnects.

When to reach for SSE vs WebSockets

Reach for SSE when you only need server-to-client push (activity feeds, notifications, log tails, AI streaming). Reach for WebSockets when you need bidirectional traffic or binary frames.

Quickstart

use futures::StreamExt;
use suprnova::{HttpResponse, Request, Response, sse::SseEvent};
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;

pub async fn stream_ticks(_req: Request) -> Response {
    let (tx, rx) = mpsc::channel::<SseEvent>(16);
    tokio::spawn(async move {
        for i in 0..10 {
            let evt = SseEvent::data(format!("tick {i}"))
                .with_event("tick")
                .with_id(i.to_string());
            if tx.send(evt).await.is_err() {
                break; // client disconnected
            }
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        }
    });
    Ok(HttpResponse::sse(ReceiverStream::new(rx)))
}

Wire output for one tick:

event: tick
id: 0
data: tick 0

The browser parses this and fires a tick event with evt.data === "tick 0" and evt.lastEventId === "0".

The SseEvent API

SseEvent is the type you push onto the stream. It has two kinds:

• Frame - a normal event with optional event / id / retry and a multi-line data payload. Built via SseEvent::data, SseEvent::json, or SseEvent::error.
• Comment - a wire-only keep-alive (:\n\n or : <text>\n\n). Built via SseEvent::comment(text) or SseEvent::keep_alive(). The browser ignores comments by spec; the bytes traversing the connection are what keep idle proxies and load balancers from closing it.

Constructors

Builders

Builders on Comment are no-ops on purpose - the wire format has no way to express "comment with an event name". A misuse stays silent rather than converting the event to a frame and surprising the producer.

Accessors

Wire encoding

SseEvent::to_wire() serializes the event to Bytes ready for the body stream:

Frame:

event: <event>\n   (only if Some)
id: <id>\n         (only if Some)
retry: <ms>\n      (only if Some)
data: <line>\n     (one per line in payload, after \r/\r\n normalization)
\n                 (terminator - required by the spec)

Comment:

: <line>\n         (one per line in comment text; `:\n` for empty lines)
\n                 (flush boundary)

Security contract

The SSE wire format uses CR / LF / NUL as field terminators with no escape mechanism. A producer that lets user input reach event: or id: without sanitizing would expose a field-injection vulnerability - a value of "legit\ndata: injected" would produce two data: fields on the wire, and "legit\n\nevent: spoofed" would terminate the current event and start a new one.

Suprnova's to_wire() defends in two layers:

• event: and id: field values - every CR / LF / NUL is stripped at serialize time. A structured WARN fires for every strip: target: "suprnova::sse", field = "event"|"id". The warn never logs the value - those bytes are attacker-controlled by construction.
• data: and comment text - \r\n and bare \r are normalized to \n before splitting, so a producer embedding \r in a payload cannot cause the receiver's parser to synthesize a data: / event: / id: field at parse time. NUL is stripped from comment text with a matching WARN.

If you want to fail fast on bad input rather than silently strip, reach for the try_with_* siblings:

use suprnova::{Response, sse::SseEvent};

let evt = SseEvent::data("hello")
    .try_with_event(&user_supplied_event)?     // returns Err on CR/LF/NUL
    .try_with_id(&user_supplied_id)?;

The returned FrameworkError::validation(field, ...) names the field; it does NOT echo the value back, so a 400 surfaced to the client is safe to log.

Keep-alive and proxy idle timeouts

Long-lived SSE connections are silent by default. Most production deployments sit behind a proxy / load balancer / CDN that closes idle connections to free resources:

• nginx default: 60 seconds
• AWS ALB default: 60 seconds
• Cloudflare default: 100 seconds

A keep_alive() comment every 15-30 seconds keeps the connection alive through all of those without dispatching a message event to the browser. The minimum-bytes form (:\n\n) is enough to flush proxy write buffers without sending any payload.

use std::time::Duration;
use futures::StreamExt;
use suprnova::sse::SseEvent;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;

let (tx, rx) = mpsc::channel::<SseEvent>(16);

// Heartbeat task - independent of the event producer.
let hb_tx = tx.clone();
tokio::spawn(async move {
    let mut ticker = tokio::time::interval(Duration::from_secs(20));
    loop {
        ticker.tick().await;
        if hb_tx.send(SseEvent::keep_alive()).await.is_err() {
            break; // client gone
        }
    }
});

// Event producer ... sends frames into `tx` as they happen.

Resume after drop (Last-Event-ID)

When the browser's EventSource drops the connection, it reconnects automatically and sends the most recent id: it saw as the Last-Event-ID header on the new request. Tag each event with .with_id(...) and read the header on the resume request:

use futures::StreamExt;
use suprnova::{HttpResponse, Request, Response, sse::{self, SseEvent}};

pub async fn stream_from_resume(req: Request) -> Response {
    let resume_from: u64 = sse::last_event_id(&req)
        .and_then(|s| s.parse().ok())
        .unwrap_or(0);

    // Build the producer stream from `resume_from + 1` onward. The closure
    // owns its own running counter so the mutation stays inside the stream.
    let stream = futures::stream::iter(events_since(resume_from))
        .scan(resume_from + 1, |next_id, payload| {
            let id = *next_id;
            *next_id += 1;
            futures::future::ready(Some((id, payload)))
        })
        .map(|(id, payload)| {
            SseEvent::json("activity", &payload)
                .expect("payload is a Serialize value")
                .with_id(id.to_string())
        });

    Ok(HttpResponse::sse(stream))
}

sse::last_event_id(&Request) -> Option<String> returns None when the header is absent OR when the value contains a NUL byte (per the WHATWG spec, NUL invalidates a last-event-id and the browser's parser would drop it). The returned String is otherwise opaque user input - parse it as your own cursor / sequence / offset before using it.

Domain-level errors

SseEvent::error("...") produces the conventional event: error\ndata: <msg>\n\n shape. Subscribers can listen for it separately from the connection-level error the browser fires on transport failure:

const es = new EventSource("/stream");

// Connection / transport errors (no `data`).
es.onerror = (evt) => console.warn("transport error", evt);

// Domain-level errors emitted by SseEvent::error(...).
es.addEventListener("error", (evt) => console.error("server-side:", evt.data));

When mapping a Stream<Item = Result<T, E>> to Stream<Item = SseEvent>, the idiomatic pattern is map(|r| match r { Ok(x) => SseEvent::json(...), Err(e) => SseEvent::error(...) }) - the consumer-side error mapping stays in the producer's hands and the framework never has to invent a default shape.

Broadcasting one stream to many subscribers

Fan-out to many SSE subscribers is already covered by the broadcasting subsystem: subscribe to a BroadcastHub channel and adapt the broadcast::Receiver into the SseEvent stream with tokio_stream::wrappers::BroadcastStream + .map(...). Each connection gets its own receiver; the hub handles slow-consumer policy (Lagged(n) errors when a subscriber falls behind) and you decide how to surface that to the client.

The working dogfood example at app/src/controllers/sse_example.rs implements this in ~25 lines:

use futures::StreamExt;
use std::sync::Arc;
use suprnova::broadcasting::BroadcastHub;
use suprnova::container::App;
use suprnova::{HttpResponse, Request, Response, sse::SseEvent};
use tokio_stream::wrappers::BroadcastStream;

pub async fn stream(_req: Request) -> Response {
    let hub: Arc<dyn BroadcastHub> = App::make::<dyn BroadcastHub>()
        .expect("BroadcastHub not bootstrapped");
    let rx = hub.subscribe("user_registered");

    let stream = BroadcastStream::new(rx).map(|result| match result {
        Ok(envelope) => SseEvent::json("user.registered", &envelope.data)
            .unwrap_or_else(|_| {
                SseEvent::data(envelope.data.to_string())
                    .with_event("user.registered")
            }),
        Err(tokio_stream::wrappers::errors::BroadcastStreamRecvError::Lagged(n)) => {
            SseEvent::data(n.to_string()).with_event("lagged")
        }
    });

    Ok(HttpResponse::sse(stream))
}

The lagged event lets the client trigger a full refetch and resume - the connection stays open through the lag.

event_stream and stream_json

HttpResponse::sse takes full control of framing - you build every SseEvent yourself. Two higher-level siblings cover the common shapes:

use suprnova::sse::{EndSignal, StreamedEvent};
use suprnova::{HttpResponse, Request, Response};
use tokio::sync::mpsc;

pub async fn progress(_req: Request) -> Response {
    let (tx, rx) = mpsc::channel::<StreamedEvent>(16);
    tokio::spawn(async move {
        for pct in [25, 50, 75, 100] {
            let evt = StreamedEvent::message(pct).unwrap();
            if tx.send(evt).await.is_err() {
                break; // client disconnected
            }
        }
    });
    let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
    Ok(HttpResponse::event_stream(stream, EndSignal::default()))
}

StreamedEvent::message(data) defaults event to "update" - what useEventStream listens for out of the box; StreamedEvent::named(event, data) overrides it for a producer fanning out more than one logical channel over the same connection. data reaches the wire unquoted for a bare string, JSON-encoded otherwise. event_stream's end: EndSignal argument controls the terminal frame sent after the stream ends: EndSignal::default() sends event: update\ndata: </stream>\n\n (Laravel's own default, and what useEventStream's endSignal option checks for); EndSignal::None omits it; EndSignal::text(...) / EndSignal::Event(...) customize it. This is Suprnova's ResponseFactory::eventStream($callback, $headers, $endStreamWith).

HttpResponse::stream_json(stream) - Laravel's ResponseFactory::streamJson / StreamedJsonResponse - takes any Stream<Item = impl Serialize> and flushes it as one incrementally-built JSON array (Content-Type: application/json) instead of buffering the whole collection first. The bytes on the wire are exactly [item,item,...]; concatenate the whole response and it deserializes with any JSON parser.

Consuming from React / Vue / Svelte

The @laravel/stream-{react,vue,svelte} packages own the client side of this wire contract - Suprnova targets theirs rather than shipping its own:

import { useEventStream, useJsonStream } from "@laravel/stream-react";

const { message } = useEventStream("/progress");          // against an event_stream endpoint
const { data, send } = useJsonStream<Order[]>("/export"); // against a stream_json endpoint

useStream/useJsonStream POST with two headers Suprnova reads like any other request header: X-STREAM-ID (a plain, non-authenticating correlation id the hook generates client-side) and X-CSRF-TOKEN, read from <meta name="csrf-token"> the same way CSRF protection already expects it. useEventStream sends neither - EventSource cannot set custom request headers at all, it's a plain browser GET.

Production setup

Response headers

HttpResponse::sse(...) sets the required headers for you:

Tuning reconnect

The default browser reconnect delay is 3 seconds. Send a retry: field once at the start of the stream to override it:

let preamble = SseEvent::data("ready").with_retry(Duration::from_secs(5));

Duration::ZERO is valid per the spec ("reconnect immediately") and is emitted verbatim - no coercion. For production streams a 5-15 second retry strikes a balance between fast recovery and not hammering the server during a regional outage.

Why Suprnova diverges

Laravel ships SSE as a one-off helper on Response: Response::eventStream(fn () => ...) takes a generator-yielding closure and frames each yielded value as a data: line. It does not model event: / id: / retry: as first-class fields, has no built-in keep-alive primitive, and does not sanitize values that would inject extra fields on the wire.

Suprnova treats SSE as a real subsystem rather than a one-off helper:

• SseEvent is a typed value with fallible (try_with_*) and infallible (with_*) builders, distinct Frame and Comment kinds, and a documented sanitization contract on every single-line field.
• HttpResponse::sse(stream) plugs into the same stream_bytes body pipeline used by any other long-lived response, so SSE shares one cancellation, headers, and panic-isolation path with the rest of the framework.
• Producers compose any Stream<Item = SseEvent> - tokio::sync::mpsc, tokio::sync::broadcast, futures::stream::iter, or the BroadcastHub fan-out adapter. None of these require a framework escape hatch.
• A Last-Event-ID reader (sse::last_event_id) and the WHATWG NUL-drop rule are in the box, so resume-after-drop is one parse call away rather than a custom header utility per app.

Reference

Next

• WebSockets - the other long-lived connection, when you need bidirectional or binary frames.
• Broadcasting - BroadcastHub fan-out shared with WebSocket subscribers.
• Notifications - channel drivers for non-streaming push delivery (mail, database, broadcast).
• Web Push - server-pushed notifications that reach the client when no EventSource is open.
• Responses - the rest of the HttpResponse builder surface.