Response Streaming Implementation Plan

[SeanTAllen]

Implementation plan for response streaming (design: #8). Request body streaming is deferred.

Scope

Add opt-in streaming response support so any handler can send chunked HTTP responses. The existing buffered response path is unchanged.

New Files

hobby/stream_sender.pony

interface tag StreamSender
  """
  Sends streaming response chunks to the client.

  Obtained from `Context.start_streaming()`. Call `send_chunk()` to send
  body data incrementally, and `finish()` to send the terminal chunk and
  complete the response.

  Each call is asynchronous — chunks are delivered in the order they are
  sent. The sender remains usable across actor boundaries; pass it to any
  actor that generates response data.
  """
  be send_chunk(data: ByteSeq)
  be finish()

No wrapper actor needed. _Connection structurally matches this interface by having send_chunk and finish behaviors. Context.start_streaming() returns _Connection tag upcast to StreamSender tag. Users can't name _Connection (package-private), so no extra capabilities leak.

Modified Files

hobby/context.pony

1. Add field let _conn: _Connection tag
2. Add field var _streaming: Bool = false
3. Update constructor to accept conn: _Connection tag parameter
4. Add start_streaming method:

fun ref start_streaming(status: stallion.Status,
  headers: (stallion.Headers val | None) = None): StreamSender tag
=>
  """
  Begin a streaming response using chunked transfer encoding.

  Sends the status line and headers immediately. Returns a `StreamSender`
  for sending body chunks asynchronously — pass it to any actor that
  generates response data. Call `send_chunk()` to send data and `finish()`
  to complete the response.

  Sets `is_handled()` to true. If a buffered response has already been
  sent, this call is silently ignored.

  Streaming requires HTTP/1.1. For HTTP/1.0 requests, this method is a
  no-op — `is_handled()` remains false, `is_streaming()` remains false,
  and the returned sender silently drops all chunks. The chain runner
  will send 500 unless the handler also calls `respond()`.
  """
  if not _handled then
    if not (request.version is stallion.HTTP10) then
      _handled = true
      _streaming = true
      _responder.start_chunked_response(status, headers)
    end
  end
  _conn

5. Add is_streaming method:

fun box is_streaming(): Bool =>
  """Returns `true` if a streaming response has been started."""
  _streaming

6. Add package-private _finish_streaming method for error cleanup:

fun ref _finish_streaming() =>
  """Finish a streaming response. Used by _ChainRunner for error cleanup."""
  if _streaming then
    _responder.finish_response()
    _streaming = false
  end

HTTP/1.0 behavior: start_streaming checks request.version. For HTTP/1.0, it skips _handled/_streaming/start_chunked_response entirely. The returned StreamSender is a no-op (because _Connection._streaming_responder is never set). The chain runner sees is_handled() == false after the handler returns and sends 500. The client gets a response rather than a hanging connection.

hobby/_connection.pony

1. Add field var _streaming_responder: (stallion.Responder | None) = None
2. Add two behaviors:

be send_chunk(data: ByteSeq) =>
  match _streaming_responder
  | let r: stallion.Responder => r.send_chunk(data)
  end

be finish() =>
  match _streaming_responder
  | let r: stallion.Responder =>
    r.finish_response()
    _streaming_responder = None
  end

3. Update on_request_complete — pass this to Context constructor and store the responder after _ChainRunner completes if streaming was started:

fun ref on_request_complete(request': stallion.Request val,
  responder: stallion.Responder)
=>
  let body: Array[U8] val =
    if _has_body then
      _has_body = false
      (_body = recover iso Array[U8] end)
    else
      recover val Array[U8] end
    end

  let path = request'.uri.path
  match _router.lookup(request'.method, path)
  | let m: _RouteMatch =>
    let ctx = Context(request', responder, m.params, body, this)
    _ChainRunner(ctx, m.handler, m.middleware)
    if ctx.is_streaming() then
      _streaming_responder = responder
    end
  else
    let response = stallion.ResponseBuilder(stallion.StatusNotFound)
      .add_header("Content-Length", "9")
      .finish_headers()
      .add_chunk("Not Found")
      .build()
    responder.respond(response)
  end

hobby/_chain_runner.pony

Update handler phase and middleware before phase to clean up partial streams on error. If a handler (or middleware) calls start_streaming() then errors before setting up its chunk-producing actor, the stream would hang. The chain runner finishes it:

Handler phase:

if not ctx.is_handled() then
  try
    handler(ctx)?
  else
    if ctx.is_streaming() then
      ctx._finish_streaming()
    end
  end
  if not ctx.is_handled() then
    ctx.respond(stallion.StatusInternalServerError, "Internal Server Error")
  end
end

Middleware before phase:

try
  m.before(ctx)?
else
  if ctx.is_streaming() then
    ctx._finish_streaming()
  elseif not ctx.is_handled() then
    ctx.respond(stallion.StatusInternalServerError, "Internal Server Error")
  end
end

hobby/_test_integration.pony

Add integration tests and register them in _TestIntegrationList.tests():

1. _TestStreamingResponse — A _StreamingTestHandler starts streaming and creates a _TestChunkProducer actor with the StreamSender. The producer sends "chunk1" and "chunk2" then calls finish(). The test client's expected string is "chunk2" (which only appears after both chunks are sent, confirming the stream was properly started and terminated — the test completes within the timeout because finish() sends the terminal chunk and closes the response cleanly).

2. _TestStreamingErrorCleanup — A handler starts streaming then raises error. The chain runner calls _finish_streaming(), which sends the terminal chunk. Verification is by test completion: if the cleanup didn't happen, the connection would hang and the test would time out. The expected string is "Transfer-Encoding: chunked" to confirm the chunked response machinery ran.

hobby/hobby.pony (package docstring)

Add a "Streaming Responses" section after the Middleware section:

## Streaming Responses

For server-sent events, large downloads, or other cases where the response
body is generated over time, use `ctx.start_streaming()`:

\```pony
primitive SSEHandler is hobby.Handler
  fun apply(ctx: hobby.Context ref) =>
    let sender = ctx.start_streaming(stallion.StatusOK)
    MyEventProducer(sender)

actor MyEventProducer
  let _sender: hobby.StreamSender tag
  new create(sender: hobby.StreamSender tag) => _sender = sender

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

  be done() =>
    _sender.finish()
\```

Middleware `after` phases run when the handler returns, which may be before
all chunks have been sent.

examples/streaming/ (new)

SSE-style example demonstrating:

• Handler calling ctx.start_streaming()
• Passing StreamSender tag to a producer actor
• Actor sending chunks via a timer and finishing

examples/README.md

Add entry for the streaming example.

CLAUDE.md

• Add stream_sender.pony to File Layout
• Add StreamSender to Public API section
• Document streaming response flow in Key Design Decisions

Release Notes

Classify as added (new feature). Create release notes file per project conventions.

Implementation Order

1. stream_sender.pony — the interface (no dependencies)
2. context.pony — add _conn field, start_streaming, is_streaming, _finish_streaming
3. _connection.pony — add streaming field, behaviors, update on_request_complete
4. _chain_runner.pony — error cleanup for partial streams
5. _test_integration.pony — streaming integration tests + register in _TestIntegrationList
6. examples/streaming/ — example program
7. examples/README.md — update
8. hobby.pony — update package docstring
9. CLAUDE.md — update architecture docs
10. Release notes

Known Limitations

One streaming response per connection at a time. _Connection stores a single _streaming_responder field. If HTTP/1.1 keep-alive delivers a second request while a streaming response to the first is still in progress, and the second request also starts streaming, the responder field is overwritten. In practice clients don't pipeline streaming requests, but this is a correctness limitation. A future enhancement could use a keyed map of responders indexed by Stallion's request ID.

Build and Test

make                  # build + run all tests
make build-examples   # verify examples compile

[SeanTAllen]

Update: start_streaming() implementation changed (PR #16)

The context.pony section above shows an implementation that checks request.version for HTTP/1.0 inside hobby and silently returns an inert sender. This approach was superseded by PR #16, which upgrades to Stallion 0.3.0 and delegates protocol detection to Stallion.

The current implementation:

fun ref start_streaming(status: stallion.Status,
  headers: (stallion.Headers val | None) = None)
  : (StreamSender tag | stallion.ChunkedNotSupported) ?
=>
  if _handled then error end
  match _responder.start_chunked_response(status, headers)
  | stallion.StreamingStarted =>
    _handled = true
    _streaming = true
    _conn
  | stallion.ChunkedNotSupported =>
    stallion.ChunkedNotSupported
  | stallion.AlreadyResponded =>
    error
  end

Key differences from the plan above:

• Partial (?): errors if _handled is already true, or if Stallion reports AlreadyResponded
• Typed return: (StreamSender tag | ChunkedNotSupported) instead of just StreamSender tag
• No HTTP/1.0 check in hobby: Stallion handles protocol detection and returns ChunkedNotSupported
• _finish_streaming() simplified: guard removed since _streaming is only set on StreamingStarted

Also, the "Known Limitations" section about one streaming response per connection was addressed by PR #15, which buffers pipelined requests during an active streaming response.

[SeanTAllen]

Streaming responses have been implemented. The architecture evolved significantly from this plan — there is no StreamSender interface, Context, _ChainRunner, or middleware system.

What was implemented:

• RequestHandler.start_streaming(status, headers) — returns StreamingStarted, ChunkedNotSupported (HTTP/1.0), or BodyNotNeeded (HEAD)
• RequestHandler.send_chunk(data) and RequestHandler.finish() for chunk delivery
• Backpressure forwarding via HandlerReceiver (throttled()/unthrottled())
• Interval-based handler timeout that works across streaming responses
• Connection state machine (_Idle → _HandlerInProgress → _Streaming) manages lifecycle
• Response interceptors run but mutations are no-ops for streaming (headers already on the wire)
• ServeFiles uses streaming for large files with backpressure support

See discussion #8's closing comment for more detail on the design differences.