Streaming and cancellation

Send a stream with Response.sendStream, and h34s writes its elements as they are produced. The request scope stays open until the stream ends, so the resources that the stream needs stay alive. When the client goes away, h34s stops the stream and waits for its cleanup.

Status. Response.sendStream and Sent are new in 0.3.0-alpha.1, in both runtimes. They replace the stream bodies of 0.2.0-alpha.1, which you returned from the handler, and ZHandler.sendStream of 0.1.0-alpha.1, which had another signature. See the migration guide.

Send a stream

Make Response.sendStream(stream) the last step of your handler, and return the Sent value that it gives. Sent is the handler's body: it tells h34s that the response is already on its way, so h34s writes nothing more.

Kyo: send a kyo.Stream[String, S], where S needs at most Async & Abort[Throwable] & Scope:

val letters: EventHandler = KHandler.define: _ =>
    Response.status(Status.Accepted)
      .andThen(Response.header("content-type", "text/plain"))
      .andThen(Response.sendStream(Stream.init(Seq("a", "b", "c"))))

ZIO: send a ZStream[Any, E, String], where E <: Throwable (or Nothing):

val letters: EventHandler = ZHandler.define(_ =>
  Response.status(Status.Accepted) *> Response.header("content-type", "text/plain") *> Response.sendStream(ZStream("a", "b", "c")),
)

Both send abc with status 202. These snippets come from the stream tests ("the guide's example sends abc with status 202" in HttpStreamSuite and HttpStreamSpec).

A returned stream does not compile. The compiler names the fix:

h34s cannot send kyo.Stream[String, Any] as a response body. Return a value with a NativeConverter,
return Unit for a 204, or end the handler with Response.sendStream(stream) to stream text or
Response.sendEvents(events) to stream server-sent events

Call sendStream once. A second call is a defect: the first stream stays the response, and the server logs the failure.

Know what is written

• The head leaves at the call. sendStream writes the head as your program set it when you call it. Over a Node socket, h34s also flushes it at once, so the client has the status and the headers before the first element. Without a socket, h34s writes the head first too (see in-process calls).
• Set the head first. Set the status, headers and cookies before you call sendStream. After the call, the head is fixed: a status, header or cookie that you set later does not reach the client, and KHandler.test and ZHandler.test show the head as it was at the call. The tests "a status, header or cookie set after sendStream is not on the wire" (HttpStreamSuite, HttpStreamSpec) prove it.
• Elements are strings. Each one is written as UTF-8 text, as is: no envelope and no separator. Node can write only strings and bytes to a response, so h34s does not accept other element types. A null element fails the stream.
• Default content type. Without a content type, h34s sends text/plain; charset=utf-8 and X-Content-Type-Options: nosniff. Set content-type yourself for another format.
• The marker. Every streamed response carries x-h34s-stream: 1, so a server-side $fetch can tell a stream from a buffered body. KHandler.$fetch.raw gives FetchResponse.Streamed for a response with this marker or a stream content type (text/event-stream, application/x-ndjson), and FetchResponse.Plain for every other response, in-process and over the network alike. A missing content-length does not make a response streamed: no in-process response has one.
• Your encoding. Encode values yourself, one string per element. For newline-delimited JSON: stream.map(value => value.toJson + "\n") (toJson comes with NativeConverter) and Response.header("content-type", "application/x-ndjson"). For server-sent events, use Response.sendEvents, which encodes the frames for you; see server-sent events.
• Chunks. HTTP clients can merge or split writes. Do not rely on one client-side chunk per element.

Validate and acquire before the call

Because the head leaves at the call, a failure after it cannot change the status. Do the work that can fail with a proper error before you call sendStream:

1. Read and validate the request body.
2. Acquire the resources that the stream needs, such as a database session or a file.
3. Set the status and the headers.
4. Call sendStream.

A failure in steps 1 to 3 uses the normal failure rule: the client gets an approved SafeError response or an opaque 500.

Keep resources alive for the transfer

Acquire what the stream needs in the request scope, then send the stream. h34s keeps the scope open until the stream ends, fails or is cancelled:

• Kyo: Scope.acquireRelease(open)(close).map(source => Response.sendStream(...stream over source...)), or use a stream that manages its own resources.
• ZIO: ZIO.acquireRelease(open)(close).flatMap(source => Response.sendStream(ZStream...)) in the handler (the request Scope is in HandlerEnv), or ZStream.unwrapScoped(...) / ZStream.acquireReleaseWith(...) for resources the stream owns.

A resource that you close before you call sendStream is gone. h34s cannot restore it.

Cancellation

The bridge is pull-driven. One producer fiber owns the stream and advances only when the HTTP side asks for the next chunk. No chunk is computed before it is requested.

When the client disconnects:

1. h34s interrupts the producer and the handler.
2. It waits for the stream's finalizers, including asynchronous ones, and for the finalizers of every scope around sendStream.
3. It closes the request scope and its resources, once, writes nothing more, and only then settles the request.

A request whose response closed before the stream started never starts the stream. Interrupted work can roll back. To make a step finish anyway, use KHandler.uninterruptible(step) (Kyo) or ZIO.uninterruptible (ZIO); see when the client disconnects.

Failures

Clients should treat an incomplete transfer as a failure, even when the status was successful.

Roll back when a stream fails

sendStream fails with the stream's error, so the code around it sees the failure. Put a transaction, or a finalizer that checks the outcome, around the step.

In these snippets, commit and rollback are your own steps. With kyo-sql, DB.transaction around the step does this for you.

Kyo: a Scope.run around the step gives its finalizers the outcome, Absent on success:

Scope.run(
  Scope.ensure((outcome: Maybe[Result.Error[Any]]) => if outcome.isEmpty then commit else rollback)
    .andThen(Response.sendStream(stream)),
)

Write that Scope.run inside your handler. A finalizer that you add directly to the request scope gets Absent even after a failure, because h34s closes the request scope after your program, without its outcome. Such a finalizer would commit. The test "after a stream failure, a finalizer of the request scope gets Absent, and one of a Scope.run around sendStream gets the failure" (HttpStreamSuite) shows the difference.

ZIO: ZIO.addFinalizerExit, inside ZIO.scoped or in the request Scope, gets the Exit (the request Scope case is the test "a stream failure after the head: a finalizer of the request Scope sees the failure" in HttpStreamSpec):

ZIO.scoped(
  ZIO.addFinalizerExit(exit => if exit.isSuccess then commit else rollback) *> Response.sendStream(stream),
)

The stream's own cause reaches your ZIO program: a ZStream[Any, MyError, String] fails sendStream with MyError. The tests "a stream failure after two elements fails the sendStream step; a Scope finalizer around it sees the failure" (HttpStreamSuite) and "a stream failure after two elements fails the sendStream effect; a finalizer of ZIO.scoped around it sees the failure" (HttpStreamSpec) prove both. A client disconnect still waits for the finalizers of such a scope: see the test "a disconnect waits for the asynchronous finalizers of a Scope.run around sendStream before the request settles" (HttpStreamSuite).

Server-sent events

Send server-sent events with Response.sendEvents(events), where events is a stream of ServerEvent values. It is Response.sendStream for the text/event-stream format: h34s sets the event-stream headers and writes one frame per event. Everything above holds for it. The head leaves at the call, the request scope stays open until the stream ends, a disconnect waits for the stream's cleanup, and a failure after the head ends the response and fails the step.

Status. Response.sendEvents, ServerEvent and $fetch.events are new in 0.3.0-alpha.1, in both runtimes.

Build the events

A ServerEvent has four fields:

ServerEvent.json(value, event, id) writes value as one line of JSON text with its NativeConverter. Read it back with $fetch.events[A] in Scala, or with JSON.parse(message.data) in the browser.

A line break inside event or id, U+0000 inside id, an empty event or a negative retry would change the meaning of the frame. The ServerEvent constructor throws an IllegalArgumentException for them, so h34s never writes such a frame. Inside a stream, the throw fails the stream like any other defect.

Send them

In these snippets, Countdown is the example's event model. Each route sends 3, 2, 1 and 0, one event per second. Kyo:

/** One event of the countdown: the seconds that remain. */
final case class Countdown(remaining: Int) derives Schema, NativeSchema

/** `GET /api/kyo/countdown`: the events 3, 2, 1 and 0, one per second, as `tick` events. A chunk of
  * one element sends each event when it is ready, not the whole chunk at its end.
  */
@JSExportTopLevel("default", "kyo-countdown")
val countdown: EventHandler = KHandler
  .define: _ =>
      val ticks = Stream.init(3 to 0 by -1, chunkSize = 1).map: remaining =>
          Async.sleep(1.second).andThen(ServerEvent.json(Countdown(remaining), event = Some("tick")))
      Response.sendEvents(ticks)
  .at(Route.get("/api/kyo/countdown"))

ZIO:

/** One event of the countdown: the seconds that remain. */
final case class Countdown(remaining: Int) derives Schema, NativeSchema

/** `GET /api/zio/countdown`: the events 3, 2, 1 and 0, one per second, as `tick` events. */
@JSExportTopLevel("default", "zio-countdown")
val countdown: EventHandler = ZHandler
  .define: _ =>
      val ticks = ZStream(3, 2, 1, 0).mapZIO: remaining =>
          ZIO.sleep(1.second).as(ServerEvent.json(Countdown(remaining), event = Some("tick")))
      Response.sendEvents(ticks)
  .at(Route.get("/api/zio/countdown"))

These are the countdown routes of the example application (examples/nuxt), and docs/test checks that the snippets equal them. In Kyo, Stream.map with an effect emits a chunk only when every element of the chunk is mapped. A one-element chunk sends each event when it is ready; with the default chunk size, all four events would leave together after four seconds.

sendEvents sets the headers of h3's createEventStream, as Response.header would:

• content-type: text/event-stream
• cache-control: private, no-cache, no-store, no-transform, must-revalidate, max-age=0
• x-accel-buffering: no, so that nginx does not buffer the stream
• connection: keep-alive, except over HTTP/2, where the header is not allowed

They replace a value of the same name that you or your handler defaults set, such as ResponseHead.noStore. Your other headers, your cookies and your status stay. Keep the status 200: a browser's EventSource fails the connection on any other status.

Each frame has an id:, event: and retry: line when the field is set, then the data: lines, then an empty line: ServerEvent.json(Countdown(3), event = Some("tick")) is event: tick\ndata: {"remaining":3}\n\n. h34s sends no heartbeat comments yet, so a proxy with an idle timeout can close a stream that stays quiet for a long time.

h34s does not wrap h3's createEventStream: in-process, it leaks two unhandled rejections per cancel, and it notices a disconnect only through an event that never fires there.

Read the events in the browser

The example reads both routes with the browser's EventSource, in its useCountdown composable:

/** One `tick` event: the JSON that `ServerEvent.json(Countdown(n))` writes. */
interface Countdown {
  remaining: number
}

type CountdownState = 'idle' | 'running' | 'done' | 'failed'

/** The countdown of `GET /api/<runtime>/countdown`, read with `EventSource`. */
export function useCountdown(runtime: 'kyo' | 'zio') {
  const remaining = ref<number | null>(null)
  const state = ref<CountdownState>('idle')
  let source: EventSource | undefined

  function stop() {
    source?.close()
    source = undefined
  }

  function start() {
    stop()
    remaining.value = null
    state.value = 'running'
    source = new EventSource(`/api/${runtime}/countdown`)
    source.addEventListener('tick', (message) => {
      const tick: Countdown = JSON.parse(message.data)
      remaining.value = tick.remaining
      if (tick.remaining === 0) {
        state.value = 'done'
        stop()
      }
    })
    // EventSource reconnects after an error. Stop instead: the run failed.
    source.addEventListener('error', () => {
      state.value = 'failed'
      stop()
    })
  }

  onScopeDispose(stop)
  return { remaining, state, start }
}

The example's countdown page (/countdown) shows one card per runtime. EventSource reconnects after an error and after the server ends the stream, so close it when you have what you need, as the composable does at 0.

The generated TypeScript type of an event route is string, the type of a sent stream. The generator does not describe the events' data yet, so the composable declares Countdown itself.

Read the events in a handler

Read another route's events with $fetch.events[E]. Each event's data is decoded as JSON for E with its NativeConverter. In Kyo, the result is a Stream[E, Async & Scope & Abort[FetchError]]:

val total: EventHandler = KHandler.define: _ =>
    for
      ticks <- KHandler.$fetch.events[Countdown]("/api/kyo/countdown")
      seen  <- ticks.run
    yield seen.map(_.remaining).sum

In ZIO, it is a ZStream[Any, FetchError, E]:

val total: EventHandler = ZHandler.define: _ =>
    for
      ticks <- ZHandler.$fetch.events[Countdown]("/api/zio/countdown")
      sum   <- ticks.runFold(0)(_ + _.remaining)
    yield sum

Both answer {"data": 6}. These snippets come from the tests "the guide's events example: a handler reads the countdown's typed events with $fetch.events" (NitroLocalFetchSuite, NitroLocalFetchSpec).

• Head first. events sends the request with accept: text/event-stream, unless you pass an accept header, and returns as soon as the head arrives, as raw does. The body lives in your Scope: when the scope closes, h34s cancels a body that you did not read to its end.
• Typed failures. A status that is not 2xx fails with FetchError.Status. A 2xx response with another content type fails with FetchError.NotEvents(status, contentType), and h34s does not read its body. The first event whose data is not JSON for E fails the stream with FetchError.Decode, which names the event by its position and type, never by its data; the events before it reach you.
• Events without a model. $fetch.events(FetchRequest(url)) gives the ServerEvent values themselves. Use it for a stream that mixes types, or for a JavaScript route that uses h3's createEventStream: such a route reaches you through its text/event-stream content type. In-process, stopping before the end of such a route's body makes h3 leak the unhandled rejections described above, so read a createEventStream route to its end.
• One frame, one event. Each ServerEvent has the fields of its own frame. Its id is the frame's own id: line; a browser's lastEventId carries over to later events, but ServerEvent.id does not.
• The parsing rules. h34s decodes the body as UTF-8 once, which removes one leading byte order mark, as a browser does. Then it reads the text with the WHATWG EventSource rules: lines end with CRLF, CR or LF, comment lines (:) are ignored, and one space after the colon is removed. A second byte order mark is part of the first line, which then names an unknown field. An event without a data: line, and an event that the stream does not end with an empty line, are not dispatched. A frame may be split anywhere between chunks.
• NDJSON. There is no typed decoder for newline-delimited JSON yet. Read the body of raw and split it into lines yourself.

NitroLocalFetchSuite and NitroLocalFetchSpec test this: "sendEvents in-process: $fetch.eventsE returns head-first and decodes the typed events as they arrive", "a JavaScript route with h3's createEventStream: the untyped $fetch.events reads its events by the content type", "events of a route that is not an event stream fails with FetchError.NotEvents; an error status with FetchError.Status", "a typed event whose data is not JSON for E fails the stream with FetchError.Decode, which names the event" and "over real HTTP (absolute URL), events split inside lines and inside a CRLF decode to the same typed events" and "events: the body is decoded as UTF-8 once, so of two leading byte order marks only the first is removed". ServerEventSuite checks the frame and the parsing rules, every split point of a text with several events, and that every event's frame decodes to the same event.

In-process calls

Nitro answers a server-side $fetch (KHandler.$fetch or ZHandler.$fetch) to your own route in-process, without a socket. A stream works there too:

• Head first. h34s writes the head, hands the stream to h3, and settles h3's handler promise at once, so the caller gets the head before the first element. With the h34s test copy of Nitro 2.13.1's local fetch, a caller had the head after 16 to 23 ms and the elements one per 200 ms after that.
• Bytes. Chunks are UTF-8 bytes, because a Web Response body accepts only Uint8Array chunks.
• Lifetime. Your program and the request scope stay open until the caller reads the body to its end or cancels it.
• A cancel waits for cleanup. A $fetch caller cancels an unread body when its Scope closes: the request scope, a Kyo Scope.run or a ZIO ZIO.scoped block. The cancel waits until the callee's stream and request finalizers have run, so the caller settles after the callee's cleanup.
• A failure after the head is typed. When the callee's stream fails after its head, the caller's read fails with FetchError.BodyFailed(status, reason): in Kyo, through Abort[FetchError] in the body's effect type; in ZIO, as the error of the body's ZStream. The reason is h34s' own text and never quotes the callee's error.
• An unread body stays open. A JavaScript caller, or a socketless Nitro preset, that drops the body without cancelling it keeps the callee's program and request scope open. h34s has no idle timeout for this case yet.
• The base URL and the context. h34s puts your app's app.baseURL in front of a path that starts with /, where Nitro mounts your routes, as Nuxt's $fetch does; a path that already starts with the base stays as it is. The call passes only the _platform and waitUntil entries of the caller's event.context, so the called route cannot read what your middleware put there, such as a session.
• Credentials stay in-process. A $fetch call to a path that starts with / gets the incoming request's headers and cookies, as in Nuxt. A call to an absolute URL gets only the headers that you pass and the content type and length of the body that you send, so the user's cookies and authorization never reach another host. Nuxt's event.$fetch forwards them to any host; h34s does not. To call another service with a credential, pass that header yourself.
• A JavaScript stream needs a stream content type. A JavaScript route that streams without the marker and without a stream content type is Plain, so raw reads it to its end, and never returns for an endless stream. Set text/event-stream or application/x-ndjson on such a route.

NitroLocalFetchSuite measures this behavior: "(b) streamed Scala route in-process: $fetch returns head-first; the body is pulled in the caller's Scope", "(c) a caller that drops a Streamed body: its Scope cancels the callee, whose finalizers run before the caller settles" and "(c) a caller that reads part of a Streamed body: the rest is cancelled when its Scope closes". The same suite shows the head arriving before the first element over real HTTP: "(b) the same streamed route over real HTTP: the head arrives before the first element". Two more tests prove the Plain rule and the credential rule: "(a) JSON route in-process: no marker and no stream content type, so Plain without content-length; asA decodes it" and "an absolute URL gets only the headers the caller passes: the incoming cookie, authorization and entity headers stay behind". The base URL, the context and the typed failure have their own tests: "app.baseURL: an in-process $fetch reaches a route under a non-root base; a path that carries the base is not prefixed again", "an in-process call passes only the caller's _platform and waitUntil: the callee cannot read the caller's other context entries" and "a callee stream that fails after its head: the caller's read fails with a typed FetchError.BodyFailed, not a panic". NitroLocalFetchSpec repeats them for ZIO callers, with "a caller that drops a Streamed body: its Scope cancels the callee and waits for the callee's finalizers (Q16)".

Test a streaming handler

KHandler.test and ZHandler.test collect what your handler sends. The result's body is TestResponse.Body.Streamed(elements); see testing.

Limits

What is tested

Kyo StreamSuite, HttpStreamSuite and NitroLocalFetchSuite, ZIO StreamSpec, HttpStreamSpec and NitroLocalFetchSpec: pull-driven delivery, the head written and flushed at the call before the first element, empty streams, order of effectful, concatenated, filtered and mapped streams, finalization exactly once after the end, cancellation waiting for asynchronous finalizers, disconnects releasing stream and request resources once (also with a Scope.run around sendStream), failures before and after the first element (the step fails, the connection closes, one log line), thrown errors and null elements, a failed write, the default content type, the socketless path (head first, UTF-8 bytes, a cancel that waits for the request, a failure that errors the caller's reads and reaches a $fetch caller as FetchError.BodyFailed), and the compile error for a returned stream. For server-sent events, ServerEventSuite adds the frame encoder and the parsing rules, and the suites above add sendEvents (headers, frames, HTTP/2, a disconnect) and $fetch.events (head first, a JavaScript createEventStream route, the typed failures, frames split over real HTTP). A real client abort against a running Nuxt server has not been tested on this branch.