eventsource-parser

import { createParser, type EventSourceMessage, type ParseError } from 'eventsource-parser'; // Simulate a readable stream, e.g., from a fetch response body async function getSimulatedSseStream(): Promise<ReadableStreamDefaultReader<Uint8Array>> { const sseData = `id: 1\nevent: message\ndata: Hello, world!\n\nretry: 5000\nid: 2\ndata: Another message\n\n`; const encoder = new TextEncoder(); const stream = new ReadableStream<Uint8Array>({ start(controller) { controller.enqueue(encoder.encode(sseData.substring(0, 20))); setTimeout(() => { controller.enqueue(encoder.encode(sseData.substring(20, 50))); }, 100); setTimeout(() => { controller.enqueue(encoder.encode(sseData.substring(50))); controller.close(); }, 200); }, }); return stream.getReader(); } async function parseSseStream() { let retryInterval = 3000; const parser = createParser({ onEvent(event: EventSourceMessage) { console.log('Received event:'); console.log(' id:', event.id || '<none>'); console.log(' event:', event.event || '<none>'); console.log(' data:', event.data); }, onRetry(interval: number) { console.log('Server requested retry interval of %dms', interval); retryInterval = interval; }, onError(error: ParseError) { console.error('Error parsing event stream:', error.message); if (error.type === 'invalid-retry') { console.error('Invalid retry value:', error.value); } } }); const sseStreamReader = await getSimulatedSseStream(); try { while (true) { const { done, value } = await sseStreamReader.read(); if (done) break; parser.feed(new TextDecoder().decode(value)); } } finally { parser.reset({ consume: true }); // Important: flush any pending data and reset state } console.log('Stream parsing complete. Last known retry interval:', retryInterval); } parseSseStream().catch(console.error);