ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: clarify queue tracking names

Browse source
This commit is contained in:
Kit Langton 2026-04-02 13:12:19 -04:00
commit 946d2eecbe
3 changed files with 9 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

8
packages/opencode/src/server/routes/event.ts
View file

@ -33,7 +33,7 @@ export const EventRoutes = () =>
c.header("X-Content-Type-Options", "nosniff") c.header("X-Content-Type-Options", "nosniff")
return streamSSE(c, async (stream) => { return streamSSE(c, async (stream) => {
const q = new AsyncQueue<string | null>({ name: "sse:event" }) const q = new AsyncQueue<string | null>({ name: "sse:event" })
let done = false let closed = false
q.push( q.push(
JSON.stringify({ JSON.stringify({
@ -53,12 +53,12 @@ export const EventRoutes = () =>
}, 10_000) }, 10_000)
const stop = () => { const stop = () => {
if (done) return if (closed) return
done = true closed = true
clearInterval(heartbeat) clearInterval(heartbeat)
unsub() unsub()
q.push(null) q.push(null)
q.done() q.untrack()
log.info("event disconnected") log.info("event disconnected")
} }

8
packages/opencode/src/server/routes/global.ts
View file

@ -20,7 +20,7 @@ export const GlobalDisposedEvent = BusEvent.define("global.disposed", z.object({
async function streamEvents(c: Context, name: string, subscribe: (q: AsyncQueue<string | null>) => () => void) { async function streamEvents(c: Context, name: string, subscribe: (q: AsyncQueue<string | null>) => () => void) {
return streamSSE(c, async (stream) => { return streamSSE(c, async (stream) => {
const q = new AsyncQueue<string | null>({ name }) const q = new AsyncQueue<string | null>({ name })
let done = false let closed = false
q.push( q.push(
JSON.stringify({ JSON.stringify({
@ -44,12 +44,12 @@ async function streamEvents(c: Context, name: string, subscribe: (q: AsyncQueue<
}, 10_000) }, 10_000)
const stop = () => { const stop = () => {
if (done) return if (closed) return
done = true closed = true
clearInterval(heartbeat) clearInterval(heartbeat)
unsub() unsub()
q.push(null) q.push(null)
q.done() q.untrack()
log.info("global event disconnected") log.info("global event disconnected")
} }

2
packages/opencode/src/util/queue.ts
View file

@ -63,7 +63,7 @@ export class AsyncQueue<T> implements AsyncIterable<T> {
}) })
} }
done() { untrack() {
if (this.id === undefined) return if (this.id === undefined) return
all.delete(this.id) all.delete(this.id)
} }