84 lines
2.1 KiB
TypeScript
84 lines
2.1 KiB
TypeScript
import { Hono } from "hono"
|
|
import { describeRoute, resolver } from "hono-openapi"
|
|
import { streamSSE } from "hono/streaming"
|
|
import { Log } from "@/util/log"
|
|
import { BusEvent } from "@/bus/bus-event"
|
|
import { Bus } from "@/bus"
|
|
import { AsyncQueue } from "../../util/queue"
|
|
|
|
const log = Log.create({ service: "server" })
|
|
|
|
export const EventRoutes = () =>
|
|
new Hono().get(
|
|
"/event",
|
|
describeRoute({
|
|
summary: "Subscribe to events",
|
|
description: "Get events",
|
|
operationId: "event.subscribe",
|
|
responses: {
|
|
200: {
|
|
description: "Event stream",
|
|
content: {
|
|
"text/event-stream": {
|
|
schema: resolver(BusEvent.payloads()),
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}),
|
|
async (c) => {
|
|
log.info("event connected")
|
|
c.header("Cache-Control", "no-cache, no-transform")
|
|
c.header("X-Accel-Buffering", "no")
|
|
c.header("X-Content-Type-Options", "nosniff")
|
|
return streamSSE(c, async (stream) => {
|
|
const q = new AsyncQueue<string | null>()
|
|
let done = false
|
|
|
|
q.push(
|
|
JSON.stringify({
|
|
type: "server.connected",
|
|
properties: {},
|
|
}),
|
|
)
|
|
|
|
// Send heartbeat every 10s to prevent stalled proxy streams.
|
|
const heartbeat = setInterval(() => {
|
|
q.push(
|
|
JSON.stringify({
|
|
type: "server.heartbeat",
|
|
properties: {},
|
|
}),
|
|
)
|
|
}, 10_000)
|
|
|
|
const stop = () => {
|
|
if (done) return
|
|
done = true
|
|
clearInterval(heartbeat)
|
|
unsub()
|
|
q.push(null)
|
|
log.info("event disconnected")
|
|
}
|
|
|
|
const unsub = Bus.subscribeAll((event) => {
|
|
q.push(JSON.stringify(event))
|
|
if (event.type === Bus.InstanceDisposed.type) {
|
|
stop()
|
|
}
|
|
})
|
|
|
|
stream.onAbort(stop)
|
|
|
|
try {
|
|
for await (const data of q) {
|
|
if (data === null) return
|
|
await stream.writeSSE({ data })
|
|
}
|
|
} finally {
|
|
stop()
|
|
}
|
|
})
|
|
},
|
|
)
|