simple sse for koa

create-sse.ts

import { Context } from 'koa'

function isPojo (obj: any): boolean {
    return obj && typeof obj === 'object' && obj.constructor === Object
}

export interface SseController {
    emit (options?: SseEmitOptions): void

    close (): void
}

export interface SseEmitOptions {
    data?: any
    retryAfter?: number
    name?: string
    id?: string | number
}

export interface SseParams {
    pingInterval?: number
}

export default function createSse (ctx: Context, params?: SseParams): SseController {
    ctx.set({
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache, no-transform',
        'Connection': 'keep-alive'
    })

    ctx.status = 200
    ctx.req.socket.setTimeout(0)
    ctx.req.socket.setNoDelay(true)
    ctx.req.socket.setKeepAlive(true)

    let interval

    const control: SseController = {
        emit (options?: SseEmitOptions): void {
            let res: string[] = []

            if (options?.name) {
                res.push(`event: ${options.name}`)
            }
            if (options?.retryAfter) {
                res.push(`retry: ${options.retryAfter}`)
            }
            if (options?.id) {
                res.push(`id: ${options.id}`)
            }

            if (isPojo(options?.data)) {
                res.push('data: ' + JSON.stringify(options?.data))
            } else if (options?.data === undefined) {
                res.push('data: ')
            } else {
                let data = String(options?.data)
                const lines = data.split(/\r\n|\r|\n/g)
                res.push(...lines.map(i => `data: ${i}`))
            }

            let str = res.join('\n') + '\n\n'
            ctx.res.write(str)
        },
        close (): void {
            if (!ctx.res.writableEnded && !(ctx.req as any).aborted) {
                control.emit({
                    name: 'closed',
                    retryAfter: 86400
                })

                ctx.res.end()
            }
            if (interval) {
                clearInterval(interval)
                interval = null
            }
        }
    }

    interval = setInterval(() => control.emit({
        name: 'heartbeat'
    }), params?.pingInterval ?? 30000)

    // idk which of those actually works, lmao
    ctx.req.on('close', control.close)
    ctx.req.on('aborted', control.close)
    ctx.req.on('error', control.close)
    ctx.res.on('error', control.close)

    return control
}

example.ts

const sleep = (s: number): Promise<void> => new Promise(resolve => setTimeout(resolve, s))

app.get('/sse', async (ctx) => {
    const control = createSse(ctx)

    for (let i = 0; i < 10; i++) {
        let wait = Math.random() * 5000
        control.emit({
            name: 'hi',
            data: { i, wait }
        })
        await sleep(wait)
    }

    control.close()
})