OpenVelo Agent
120 lines
3.4 KiB
TypeScript
120 lines
3.4 KiB
TypeScript
import { EventSourceLike } from './event-status.pure';
|
|
|
|
export function isBootstrapGeneralFrame(frame: unknown): frame is { topic: 'general'; themeKey?: string; registrationsEnabled?: boolean } {
|
|
if (!frame || typeof frame !== 'object') return false;
|
|
const f = frame as Record<string, unknown>;
|
|
return f['topic'] === 'general';
|
|
}
|
|
|
|
export type BootstrapGeneralPayload = { topic: 'general'; themeKey?: string; registrationsEnabled?: boolean };
|
|
|
|
export function parseSseDataLines(raw: string): string | null {
|
|
let event = 'message';
|
|
const dataLines: string[] = [];
|
|
for (const line of raw.split('\n')) {
|
|
if (line.startsWith(':')) continue;
|
|
if (line.startsWith('event:')) event = line.slice(6).trim();
|
|
else if (line.startsWith('data:')) dataLines.push(line.slice(5).trim());
|
|
}
|
|
if (dataLines.length === 0) return null;
|
|
return dataLines.join('\n');
|
|
}
|
|
|
|
export function makeBootstrapEventSource(url: string, fetchImpl: typeof globalThis.fetch = fetch): EventSourceLike {
|
|
const listeners = new Map<string, Array<(ev: MessageEvent | Event) => void>>();
|
|
let controller: AbortController | null = new AbortController();
|
|
let closed = false;
|
|
let buffered: string[] = [];
|
|
|
|
const dispatch = (type: string, ev: MessageEvent | Event) => {
|
|
const arr = listeners.get(type);
|
|
if (!arr) return;
|
|
for (const fn of arr) {
|
|
try {
|
|
fn(ev);
|
|
} catch {
|
|
// ignore listener errors
|
|
}
|
|
}
|
|
};
|
|
|
|
const flushBuffered = () => {
|
|
if (closed) return;
|
|
const items = buffered;
|
|
buffered = [];
|
|
for (const data of items) {
|
|
dispatch('message', new MessageEvent('message', { data }));
|
|
}
|
|
};
|
|
|
|
const fireError = () => {
|
|
if (closed) return;
|
|
dispatch('error', new Event('error'));
|
|
};
|
|
|
|
const start = async () => {
|
|
try {
|
|
const res = await fetchImpl(url, {
|
|
method: 'GET',
|
|
headers: { Accept: 'text/event-stream' },
|
|
credentials: 'include',
|
|
signal: controller!.signal,
|
|
});
|
|
if (!res.ok || !(res as unknown as { body?: unknown }).body) {
|
|
fireError();
|
|
return;
|
|
}
|
|
const body = (res as unknown as { body: ReadableStream<Uint8Array> }).body;
|
|
const reader = body.getReader();
|
|
const decoder = new TextDecoder();
|
|
let buf = '';
|
|
for (;;) {
|
|
const { value, done } = await reader.read();
|
|
if (done) break;
|
|
buf += decoder.decode(value, { stream: true });
|
|
let idx: number;
|
|
while ((idx = buf.indexOf('\n\n')) >= 0) {
|
|
const raw = buf.slice(0, idx);
|
|
buf = buf.slice(idx + 2);
|
|
const payload = parseSseDataLines(raw);
|
|
if (payload === null) continue;
|
|
if (listeners.has('message')) {
|
|
dispatch('message', new MessageEvent('message', { data: payload }));
|
|
} else {
|
|
buffered.push(payload);
|
|
}
|
|
}
|
|
}
|
|
} catch {
|
|
if (!closed) fireError();
|
|
}
|
|
};
|
|
|
|
void start();
|
|
|
|
return {
|
|
addEventListener(type, listener) {
|
|
const arr = listeners.get(type) ?? [];
|
|
arr.push(listener);
|
|
listeners.set(type, arr);
|
|
if (type === 'message' && buffered.length > 0) {
|
|
queueMicrotask(flushBuffered);
|
|
}
|
|
},
|
|
close() {
|
|
if (closed) return;
|
|
closed = true;
|
|
if (controller) {
|
|
try {
|
|
controller.abort();
|
|
} catch {
|
|
// ignore
|
|
}
|
|
controller = null;
|
|
}
|
|
listeners.clear();
|
|
buffered = [];
|
|
},
|
|
};
|
|
}
|