ServerSentEventsStream Koa Example

ServerSentEventsStream.ts

import { TransformStream } from "stream/web"

export default class ServerSentEventsStream extends TransformStream<unknown, string> {
  constructor() {
    super({
      async transform(message, controller) {
        controller.enqueue(`data: ${JSON.stringify(message)}` + "\n\n")
      },
    })
  }
}

usage.ts

import { Readable } from "stream"
import ServerSentEventsStream from "./ServerSentEventsStream"

function sendResponse(ctx: KoaContext) {
  ctx.setHeader("Content-Type", "text/event-stream")
  ctx.setHeader("Cache-Control", "no-cache")
  ctx.setHeader("X-Accel-Buffering", "no")
  ctx.body = Readable.fromWeb(
    getEventsStream()
      .pipeThrough(new ServerSentEventsStream())
      .pipeThrough(new TextEncoderStream())
  )
}