Streaming Design

[SeanTAllen]

Two independent, composable features: streaming responses and streaming request bodies. Response streaming is the priority for initial implementation; request body streaming is designed here for future reference.

1. Streaming Responses

Any handler can send a chunked HTTP response. Stallion already has start_chunked_response(), send_chunk(), and finish_response() on Responder.

Public API

New interface:

interface tag StreamSender
  """Send response chunks and signal completion."""
  be send_chunk(data: ByteSeq)
  be finish()

Context gains one method:

fun ref start_streaming(status: Status,
  headers: (Headers val | None) = None): StreamSender tag

This calls _responder.start_chunked_response() synchronously (we're inside _Connection's callback), sets _handled = true, and returns _Connection tag upcast to StreamSender tag. No wrapper actor, no extra message hop — _Connection structurally matches StreamSender by having send_chunk and finish behaviors that forward to the stored Responder.

Internal Changes

Context:

• New field: _conn: _Connection tag (passed as this from _Connection)
• New field: var _streaming: Bool = false
• New method: start_streaming() — calls _responder.start_chunked_response(), sets _handled and _streaming, returns _conn as StreamSender tag
• New method: is_streaming(): Bool — read-only (box)
• New package-private method: _finish_streaming() — calls _responder.finish_response() for error cleanup (see "Error Handling for Partial Streaming")

_Connection:

• New field: var _streaming_responder: (Responder | None) = None
• Two new behaviors: send_chunk and finish (matching StreamSender)
• After _ChainRunner returns, check ctx.is_streaming() — if true, store the responder for later behavior turns

_ChainRunner:

• Track whether the handler errored. If it errored AND ctx.is_streaming(), call ctx._finish_streaming() to send the terminal chunk and prevent a hung connection (see "Error Handling for Partial Streaming")

Usage

primitive SSEHandler is Handler
  fun apply(ctx: Context ref) =>
    let sender = ctx.start_streaming(StatusOK)
    MySSEProducer(sender)

actor MySSEProducer
  let _sender: StreamSender tag
  new create(sender: StreamSender tag) => _sender = sender

  be emit(data: String) =>
    _sender.send_chunk("data: " + data + "\n\n")

  be done() =>
    _sender.finish()

Why This Works with Capabilities

_Connection is package-private, so users can't name its type — they only interact through StreamSender tag. The only public behaviors on _Connection are send_chunk and finish (everything from lori/stallion traits is _-prefixed or fun ref). No extra capabilities leak.

2. Streaming Request Bodies (Future)

Routes opt in to receive body chunks incrementally instead of buffering the complete body.

Public API

New interfaces:

interface val StreamingHandler
  """Handler for routes with streaming request bodies.
  Called when headers arrive (before body). Return a BodyReceiver
  to process chunks as they arrive."""
  fun apply(ctx: Context ref): BodyReceiver tag ?

interface tag BodyReceiver
  """Receives request body chunks incrementally."""
  be chunk(data: Array[U8] val)
  be finished()

Both Handler and StreamingHandler are interface val — both can be primitives or classes. The handler is stateless and shareable. The difference is what the handler creates: a StreamingHandler creates an actor (the BodyReceiver) to process body chunks over time, and returns it.

Route registration — separate methods:

// On Application:
fun ref post_streaming(path: String, handler: StreamingHandler,
  middleware: (Array[Middleware val] val | None) = None)

One *_streaming variant per HTTP method. Internally, _RouteDefinition gains a streaming: Bool field, and _RouteMatch stores (Handler | StreamingHandler).

Internal Changes

For streaming routes, the handler is invoked in on_request() (when headers arrive) instead of on_request_complete():

on_request() → route lookup → if streaming route:
  1. Run middleware chain + StreamingHandler
  2. Handler returns BodyReceiver tag, stored on _Connection
  3. Middleware after phases run

on_body_chunk() → if streaming: forward to BodyReceiver.chunk()
                   else: accumulate as before

on_request_complete() → if streaming: call BodyReceiver.finished()
                         else: existing buffered logic

Non-streaming routes are completely unchanged — on_request() is a no-op for them.

Usage

class val FileUploadHandler is StreamingHandler
  fun apply(ctx: Context ref): BodyReceiver tag =>
    let sender = ctx.start_streaming(StatusOK)  // response streaming too
    FileUploadProcessor(sender)

actor FileUploadProcessor is BodyReceiver
  let _sender: StreamSender tag

  new create(sender: StreamSender tag) =>
    _sender = sender

  be chunk(data: Array[U8] val) =>
    // Write to file, process, etc.

  be finished() =>
    _sender.send_chunk("Upload complete")
    _sender.finish()

StreamingHandler.apply() Returns BodyReceiver tag (Not Set via Context)

The return type makes the contract explicit — you MUST provide a receiver. And if the handler errors, no receiver is set (error prevents the return), so chunks are safely dropped and 500 is sent. With a Context setter, the receiver could be set before an error, leaving state partially initialized.

3. Composition

Response streaming and request body streaming compose naturally. A StreamingHandler can call ctx.start_streaming() to get a StreamSender, then return a BodyReceiver that forwards processed data through the sender. Both features use the same StreamSender interface.

Response streaming is also available on non-streaming routes — any regular Handler can call ctx.start_streaming().

4. Design Decisions

Middleware Interaction Is Unchanged

before/after phases run around the handler setup, not around the full streaming lifecycle. For streaming body routes, after runs after the handler returns (receiver is set up but body hasn't arrived). This is consistent — after is for request-level cleanup/logging, not streaming lifecycle management.

HTTP/1.0 Limitation

Stallion silently rejects chunked encoding for HTTP/1.0. start_streaming() always returns a StreamSender, but chunks are silently dropped for HTTP/1.0 clients. This is documented but not expressed in the type (HTTP/1.0 is vanishingly rare).

Flow Control Is Deferred

Stallion has on_chunk_sent(token) and on_throttled()/on_unthrottled() callbacks. Exposing these through StreamSender is a natural future extension but not included initially.

Error Handling for Partial Streaming

If a handler calls start_streaming() and then errors, the chunked response has already started (status line and headers are on the wire) but nobody will call finish(). Since _handled is true, _ChainRunner would normally skip the 500 — leaving the connection hung with an unterminated chunked response.

To prevent this, _ChainRunner tracks whether the handler errored. If it errored AND the context is in streaming mode, it finishes the stream (sends the terminal chunk via a package-private Context._finish_streaming() method that calls _responder.finish_response()). This applies to both response-only streaming and streaming body routes (where the handler errors before returning a BodyReceiver).

Streaming Body Responses Always Use Chunked Transfer Encoding

If a streaming body handler wants to send a deferred response after body processing, it starts streaming and sends a single chunk + finish. This is valid HTTP/1.1 and avoids needing a separate deferred-response mechanism. Content-Length responses require the non-streaming path.

Separate *_streaming() Methods vs. Parameter

Separate methods take StreamingHandler instead of Handler, making a type mismatch a compile error. A parameter approach would need a runtime check.

5. What Changes, What Doesn't

Component	Changes?	Details
Handler	No	Unchanged
Middleware	No	Unchanged
Context	Yes	Add start_streaming(), is_streaming(), _finish_streaming(), _conn field, _streaming flag
Application	Yes	Add *_streaming() route methods (request body streaming only)
_Connection	Yes	Add streaming behaviors, streaming state; on_request() override (request body streaming only)
_ChainRunner	Yes	Error cleanup for abandoned streams (response streaming); handle (Handler | StreamingHandler) union (request body streaming only)
_RouteDefinition	Yes	Add streaming flag (request body streaming only)
_RouteMatch	Yes	Store (Handler | StreamingHandler) (request body streaming only)
_Router	Minor	Pass through streaming flag (request body streaming only)
StreamSender	New	Interface tag — response streaming
StreamingHandler	New	Interface val — request body streaming
BodyReceiver	New	Interface tag — request body streaming

[SeanTAllen]

Section 1 (Streaming Responses) implemented in #14. Known limitations deferred to separate issues:

• Return typed result from start_streaming() #12: start_streaming() silently ignores HTTP/1.0 and already-responded cases
• Buffer pipelined requests during streaming responses #13: Pipelined requests during streaming aren't buffered yet

Section 2 (Request Body Streaming) remains for future work.

[SeanTAllen]

Update: start_streaming() now returns a typed result (PR #16)

The "HTTP/1.0 Limitation" section above stated that start_streaming() always returns a StreamSender and silently drops chunks for HTTP/1.0 clients, with the limitation "not expressed in the type." This is no longer accurate.

Stallion 0.3.0 (ponylang/stallion#52) changed start_chunked_response() to return StartChunkedResponseResult — a union of StreamingStarted | AlreadyResponded | ChunkedNotSupported. PR #16 upgrades hobby to Stallion 0.3.0 and surfaces this through Context.start_streaming():

fun ref start_streaming(status: stallion.Status,
  headers: (stallion.Headers val | None) = None)
  : (StreamSender tag | stallion.ChunkedNotSupported) ?

• StreamingStarted → returns StreamSender tag (same as before)
• ChunkedNotSupported → returned as data so the handler can fall back to ctx.respond()
• AlreadyResponded → maps to error (programmer mistake, defensive backstop behind hobby's _handled guard)

The Context method signature section and the usage examples above show the old non-partial signature returning only StreamSender tag. The current API requires callers to match on the result:

primitive StreamHandler is Handler
  fun apply(ctx: Context ref) ? =>
    match ctx.start_streaming(stallion.StatusOK)?
    | let sender: StreamSender tag =>
      MyProducer(sender)
    | stallion.ChunkedNotSupported =>
      ctx.respond(stallion.StatusOK, "Chunked encoding not supported.")
    end

[SeanTAllen]

Streaming responses have been implemented, though the design evolved significantly from what was sketched here.

What was implemented:

• RequestHandler.start_streaming(status, headers) begins a chunked transfer-encoded response, returning StreamingStarted, ChunkedNotSupported (HTTP/1.0 fallback), or BodyNotNeeded (HEAD requests)
• RequestHandler.send_chunk(data) sends a chunk
• RequestHandler.finish() sends the terminal chunk
• Backpressure forwarding via HandlerReceiver — throttled()/unthrottled() propagate from the connection
• Interval-based handler timeout that checks a timestamp rather than cancel/recreate per chunk
• Response interceptors run but mutations are no-ops for streaming (headers already on the wire)
• ServeFiles uses streaming for large files via _FileStreamer with backpressure support

Key differences from this design:

• No separate StreamSender interface — streaming methods live on RequestHandler, which handler actors embed directly
• The actor-per-request model (HandlerFactory returning HandlerReceiver tag) replaced the Handler primitive + Context design entirely
• Connection state machine (_Idle → _HandlerInProgress → _Streaming) manages the lifecycle

Not implemented:

• Streaming request bodies — tracked in Streaming request bodies #87