Server-sent events

An operation that returns text/event-stream streams its response instead of resolving to one body. Kubb generates a function for it that hands back a typed event stream you read with for await, rather than the usual RequestResult. Events are typed from the spec, and the native response stays reachable alongside them.

Both @kubb/plugin-fetch and @kubb/plugin-axios support this. On axios, streaming needs the fetch adapter, which Kubb selects for a stream request unless you set an adapter yourself.

Consume a stream ​

A streaming operation returns an EventStreamResult: a stream to iterate and the native response. Loop over the stream with for await and read each event's data:

import { streamEvents } from './gen/clients/streamEvents'

const { stream } = await streamEvents({})

for await (const event of stream) {
  console.info(event.data)
}

data is typed from the operation's response schema. Each event also carries the optional SSE fields, so you can branch on the event name or read the id:

type ServerSentEvent<TData> = {
  data: TData
  event?: string
  id?: string
  retry?: number
}

The runtime parses each event's data as JSON when it is valid and keeps it as the raw string otherwise.

Stop early ​

Break out of the loop to stop reading. The underlying reader is canceled when the iterator finishes, so an early break releases the stream:

const { stream } = await streamEvents({})

for await (const event of stream) {
  if (event.event === 'done') break
  render(event.data)
}

To stop a stream from outside the loop, pass an AbortSignal and abort it. Aborting ends the request and the iteration:

const controller = new AbortController()
const { stream } = await streamEvents({ signal: controller.signal })

// elsewhere, for example on component unmount
controller.abort()

Read the response ​

The native response sits next to the stream, so status and headers are there before you start reading events. Check it for a header the server sets when the stream opens:

const { stream, response } = await streamEvents({})

console.info(response.status) // 200
console.info(response.headers.get('x-stream-id'))

for await (const event of stream) {
  handle(event.data)
}

Handle errors ​

A stream that fails to open rejects before you reach the loop, so the await that starts it is where a failed handshake surfaces. Wrap it to catch a non-2xx or a connection that never established:

import { ResponseError } from './gen/.kubb/client'

try {
  const { stream } = await streamEvents({})
  for await (const event of stream) {
    handle(event.data)
  }
} catch (error) {
  if (ResponseError.is(error)) {
    console.error('stream rejected', error.status)
  }
}

A drop mid-stream ends the for await loop. Track the last event's id if you need to reconnect from where the stream stopped.

See also ​

• Call operations
• Error handling
• @kubb/plugin-fetch
• @kubb/plugin-axios