Streaming responses

Return a chunked body or a Server-Sent Events stream from a handler — poem's response types pass straight through.

A handler can return a body that streams instead of one buffered whole in memory, or a live Server-Sent Events feed the browser reads with EventSource. A streamed body is any poem response type, passed through untouched — no wrapper, no schema requirement; an events feed is its own decorator, #[sse].

#[get("/download")]
async fn download(&self, query: Valid<Query<TranscodeDto>>) -> Result<Response> {
    let file = query.into_inner().file;
    match self.svc.open_result(&file).await.map_err(internal)? {
        Some(stream) => Ok(Response::builder()
            .content_type("audio/mpeg")
            .body(Body::from_bytes_stream(stream))),
        None => Err(Error::from_status(StatusCode::NOT_FOUND)),
    }
}

Body::from_bytes_stream is the whole trick: the object’s bytes flow chunk by chunk, never whole in memory.

Streamed bodies

poem::Body::from_bytes_stream turns any Stream<Item = Result<Bytes, E>> into a response body. Feed it a source that itself streams — the object-storage client’s get_stream hands back the S3 GetObject chunks directly, so a large download proxies through the server without ever sitting whole in process memory.

Server-Sent Events

#[sse("/path")] is a GET that answers a long-lived text/event-stream. The handler returns an SseStream of SseEvents and nothing else — the decorator owns the response, the keep-alive and the connection ceiling.

#[sse("/events")]
async fn events(&self, query: Valid<Query<TranscodeDto>>) -> SseStream {
    let stream = /* a Stream<Item = SseEvent> polling the job */;
    SseStream::new(stream)
}

SseEvent and SseStream come from nest_rs::http, which also re-exports futures_util — so a controller that streams declares no transport crate of its own. SseStream::new is what makes the stream 'static: an async fn on &self returning impl Stream captures the &self lifetime, and a response body has to outlive the call.

The OpenAPI document types the route from the decorator — text/event-stream, with no annotation. A hand-built streamed Response states its own with #[api(response_content_type = "audio/mpeg")]; on an #[sse] route that key is a compile error, because it could only describe something the route never sends.

Configuration

Two NESTRS_HTTP__* keys, both defaulting to something usable:

Key	Default	What it does
`SSE_MAX_CONNECTION_SECS`	`14400` (4 h)	ends the stream, so the client’s `EventSource` reconnects. `0` ⇒ unlimited
`SSE_KEEP_ALIVE_SECS`	`15`	comment interval that keeps an idle feed alive through proxies. `0` ⇒ none

A stream runs inside the request that opened it, even though its handler returned long before the first event. current_trace_id() and current_actor_id() answer there, and every event the stream logs carries the request’s trace_id — see Correlation. That is the framework’s doing, not the decorator’s: the same holds for any streaming body a handler returns.

The ceiling is a security control, and the same one NESTRS_WS__MAX_MESSAGE_BYTES bounds an inbound frame at the protocol layer — 64 KiB by default, so buffering stops before a giant frame is fully read. NESTRS_WS__MAX_CONNECTION_SECS and NESTRS_GRAPHQL__MAX_CONNECTION_SECS carry, at the same default. A stream is authenticated once, when the request arrives, then emits with those privileges for as long as it lives — so without a ceiling it outlives an expired token, a logout, or a revoked grant. Reconnecting re-runs the guard chain.

It bounds emission: the deadline is evaluated whenever the response body is polled, so no event is produced past the ceiling at any rate the client reads at. Reconnecting re-runs the guard chain, which is the window that matters.

What it does not bound is the socket, and that is reported rather than quietly closed. A peer that stops reading parks the write, hyper stops polling the body, and the connection — its task, its buffers, everything the stream holds — outlives the ceiling until that peer reads again. Two attempts to close it inside the framework were both wrong, in opposite directions, and both looked right: the only thing still polled while a write is parked is the socket, and a socket does not know which response its bytes belong to. Bounding it there truncated unrelated traffic — a full-speed client downloading 4 MB over a connection that had carried a stream received 1.4 MB under a declared content-length — and HTTP/2 multiplexes an origin’s whole traffic onto that one socket.

Bound idle sockets at the server or the reverse proxy until the transport can express “this response is stalled”: that is the shape enforceable without knowing whose bytes are queued.

Gate a stream with a capability-only guard (#[use_guards(TranscodeGuard)]), never #[authorize]: the posture masks the response against the entity model, and an event stream is no wire model to reconcile. That combination is a compile error saying so.

Limits

A streamed body has no schema in the document, only a media type and OpenAPI’s string / format: binary — there is no Json<T> to derive one from.
SSE and streamed bodies are HTTP only; the WebSocket gateway is the push channel for two-way traffic.
Compression sits outside the handler, so a streamed body is still encoded if the client accepts it — usually you leave compression off for already-compressed media.

Going further

File storage — the streaming object-store client.
audio in the demo — the streamed download and SSE feed in full.
File uploads — the multipart counterpart.
Correlation — what a stream’s own events are filed under.