85 lines
2.0 KiB
TypeScript
85 lines
2.0 KiB
TypeScript
import type { AppEnv } from '@/types/hono';
|
|
import { Hono } from 'hono';
|
|
|
|
type EventCallback = (data: string) => void;
|
|
|
|
class SSEManager {
|
|
private channels: Map<string, Set<EventCallback>> = new Map();
|
|
|
|
subscribe(channel: string, callback: EventCallback): () => void {
|
|
if (!this.channels.has(channel)) {
|
|
this.channels.set(channel, new Set());
|
|
}
|
|
this.channels.get(channel)!.add(callback);
|
|
|
|
return () => {
|
|
const subs = this.channels.get(channel);
|
|
if (subs) {
|
|
subs.delete(callback);
|
|
if (subs.size === 0) {
|
|
this.channels.delete(channel);
|
|
}
|
|
}
|
|
};
|
|
}
|
|
|
|
emit(channel: string, data: string): void {
|
|
const subs = this.channels.get(channel);
|
|
if (subs) {
|
|
for (const callback of subs) {
|
|
callback(data);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
export const sseManager = new SSEManager();
|
|
|
|
export function registerEventsRoutes(app: Hono<AppEnv>) {
|
|
app.get('/api/events/:channel', async (c) => {
|
|
const channel = c.req.param('channel');
|
|
|
|
const stream = new ReadableStream({
|
|
start(controller) {
|
|
const encoder = new TextEncoder();
|
|
|
|
const send = (data: string) => {
|
|
try {
|
|
controller.enqueue(encoder.encode(`data: ${data}\n\n`));
|
|
} catch {
|
|
// Controller closed
|
|
}
|
|
};
|
|
|
|
send('connected');
|
|
|
|
const unsubscribe = sseManager.subscribe(channel, send);
|
|
|
|
const keepAliveInterval = setInterval(() => {
|
|
send('ping');
|
|
}, 30000);
|
|
|
|
c.req.raw.signal.addEventListener('abort', () => {
|
|
console.log(`[SSE] Connection closed for channel: ${channel}`);
|
|
clearInterval(keepAliveInterval);
|
|
unsubscribe();
|
|
try {
|
|
controller.close();
|
|
} catch {
|
|
// Already closed
|
|
}
|
|
});
|
|
},
|
|
});
|
|
|
|
return new Response(stream, {
|
|
headers: {
|
|
'Content-Type': 'text/event-stream',
|
|
'Cache-Control': 'no-cache',
|
|
Connection: 'keep-alive',
|
|
'X-Accel-Buffering': 'no',
|
|
},
|
|
});
|
|
});
|
|
}
|