Server-Sent Events — `nestrs_core::sse`

Stream events to the browser with SseResponse<S>, IntoSseEvent, and serialize_to_event. One wrapper, no axum dep at the handler site.

nestrs-core ships a thin Server-Sent Events wrapper behind the sse feature. The intent is the same as the nestrs-oauth2::cookies surface: re-export the axum primitive (Sse<S>), give it a stable, axum-agnostic return type (SseResponse<S>), and add a conversion surface that covers the payload shapes handler authors actually reach for — &str, String, Bytes, and anything Serialize via serialize_to_event. Handlers can stream events without ever naming axum::response::sse.

Enable the feature

On the umbrella crate:

[dependencies]
nestrs = { version = "1.3.0", features = ["sse"] }

Or on nestrs-core directly:

[dependencies]
nestrs-core = { version = "1.3.0", features = ["sse"] }

The flag pulls in bytes, futures-core, and serde on top of the axum SSE plumbing, which is on by default in axum 0.7. Enable it only if you actually emit SSE streams; outbound clients that consume SSE payloads don’t need this flag.

Return an SSE stream

axum 0.7’s Sse::new takes Stream<Item = Result<Event, E>>. Both from_stream and from_fallible_stream are that constructor — there is no infallible-item overload.

use axum::response::sse::{Event, KeepAlive};
use futures_core::Stream;
use nestrs_core::sse::{IntoSseEvent, SseResponse};

async fn tick_handler() -> SseResponse<impl Stream<Item = Result<Event, std::convert::Infallible>>> {
    let stream = futures_core::stream::iter([
        "tick".into_sse_event(),
    ]);
    SseResponse::from_stream(stream).keep_alive(KeepAlive::new())
}

SseResponse<S> is a newtype around axum::response::sse::Sse<S> that implements IntoResponse directly, so handlers can return it without naming axum’s SSE type. The wrapper delegates the response conversion to axum, so behavior (the text/event-stream content-type, chunked transfer encoding, retry semantics) stays identical.

IntoSseEvent — payload shapes

The trait covers the shapes that convert without going through serde. JSON payloads use the free function serialize_to_event instead of a blanket T: Serialize impl — axum::response::sse::Event itself implements Serialize, so a blanket impl would conflict with the passthrough.

use nestrs_core::sse::IntoSseEvent;

let evt = "hello".into_sse_event().unwrap();

Payload	Event name	Notes
&str / String	(default — no event: field)	Plain data: line, e.g. data: hello
Bytes	(default)	UTF-8 decoded into a data: line; invalid UTF-8 is an axum::Error
Event	(passthrough)	Already an event; returned as-is

serialize_to_event

Any Serialize type becomes an SSE event with the default message name. Serialization failures (e.g. a f64::NAN that serde_json rejects) are not a stream crash — the helper emits an event: error event with the failure message instead:

use nestrs_core::sse::serialize_to_event;
use serde::Serialize;

#[derive(Serialize)]
struct Tick {
    id: u32,
    text: String,
}

// Renders as `event: message\ndata: {"id":1,"text":"hi"}\n\n`.
let evt = serialize_to_event(&Tick { id: 1, text: "hi".into() }).unwrap();

// serde_json rejects NaN — the stream does NOT die.
let _err_evt = serialize_to_event(&f64::NAN).unwrap();

axum’s Event fields are private — assert on the serialized body (data: … / event: error) rather than evt.event.

This is a deliberately conservative default — SSE consumers typically expect a long-lived stream and treat any termination as an implicit error. If you’d rather crash the stream on serialization failure, build the Event yourself with serde_json::to_string(...) and surface the Err from your stream directly.

Fallible streams

For producers that want a structured error path (a network drop, a serialization failure the caller does want to crash on), pass a Result stream. from_fallible_stream is an alias of from_stream so call sites that already produce Result read as intent:

use axum::Error as AxumError;
use axum::response::sse::Event;
use futures_core::Stream;
use nestrs_core::sse::SseResponse;

fn handler() -> SseResponse<impl Stream<Item = Result<Event, AxumError>>> {
    let stream = futures_core::stream::iter(vec![
        Ok(Event::default().data("ok")),
        Err(AxumError::new("simulated")),
    ]);
    SseResponse::from_fallible_stream(stream)
}

Result<Event, E> is axum’s contract — E must convert to Box<dyn Error + Send + Sync>.

KeepAlive

Attach a heartbeat policy with .keep_alive(...). The default KeepAlive::new() sends a comment line every 15 seconds — long enough to keep middlebox connections warm, short enough that the consumer sees activity when the producer goes quiet:

use axum::response::sse::KeepAlive;
use nestrs_core::sse::SseResponse;

let response = SseResponse::from_stream(stream)
    .keep_alive(KeepAlive::new());

Conversion from an existing Sse

If you’ve already constructed an axum::response::sse::Sse<S> (e.g. from third-party code), wrap it via From:

use axum::response::sse::Sse;
use nestrs_core::sse::SseResponse;

let sse: Sse<MyStream> = Sse::new(stream);
let response: SseResponse<MyStream> = SseResponse::from(sse);

Accessors

When the wrapper hides something you need (a custom header, a non-standard content-type), drop down to the inner Sse<S> with into_inner or borrow it with as_inner:

let inner: Sse<MyStream> = wrapper.into_inner();
let inner_ref: &Sse<MyStream> = wrapper.as_inner();

Why these choices

• Stable return type — handlers return SseResponse<S> regardless of which crate (nestrs, nestrs-http) wired the response. Tests can construct a response directly without depending on the full runtime.
• IntoSseEvent plus serialize_to_event — strings and bytes stay on the trait; JSON goes through a free function so Event’s own Serialize impl can still pass through as an event.
• Serialize-failure → error event — SSE consumers treat stream termination as a connection drop, so silently emitting an error event keeps the stream alive and lets the consumer decide what to do. Crashing the producer would be the more disruptive default.
• Behind a feature flag — sse is off by default. Apps that emit SSE streams opt in; apps that consume SSE payloads from upstream services stay on the default feature set.

See also

• axum::response::sse — the underlying primitive.
• mintlify-docs/concepts/middleware-pipeline — for per-request hooks that emit progress events to an SSE stream.
• mintlify-docs/concepts/controllers — for the handler wiring that returns SseResponse<S>.