A Server-Sent Events Consumer That Reconnects

// SSE wire format: lines separated by \n. Each event is a block separated by
// a blank line. Fields: `data:`, `event:`, `id:`, `retry:`. Anything else is ignored.
// Stage 1 of 2: a generator that yields events from an async chunk source.

async function* parseSse(chunkIter) {
    let buf = '';
    let event = { type: 'message', data: '', id: '', retry: null };
    for await (const chunk of chunkIter) {
        buf += chunk;
        let nl;
        while ((nl = buf.indexOf('\n')) !== -1) {
            const line = buf.slice(0, nl).replace(/\r$/, '');
            buf = buf.slice(nl + 1);
            if (line === '') {
                if (event.data) yield { ...event, data: event.data.replace(/\n$/, '') };
                event = { type: 'message', data: '', id: event.id, retry: event.retry };
                continue;
            }
            if (line.startsWith(':')) continue; // comment
            const colon = line.indexOf(':');
            const field = colon === -1 ? line : line.slice(0, colon);
            let value = colon === -1 ? '' : line.slice(colon + 1);
            if (value.startsWith(' ')) value = value.slice(1);
            if (field === 'data') event.data += value + '\n';
            else if (field === 'event') event.type = value;
            else if (field === 'id') event.id = value;
            else if (field === 'retry') event.retry = parseInt(value, 10);
        }
    }
    if (event.data) yield { ...event, data: event.data.replace(/\n$/, '') };
}

// Drive the parser with a fake stream so this stage runs standalone.
async function* fakeStream() {
    yield 'id: 1\nevent: progress\ndata: 10\n\n';
    yield 'id: 2\ndata: line one\ndata: line two\n\n';
    yield ': keepalive\n\n';
    yield 'retry: 2000\nid: 3\ndata: done\n\n';
}

(async () => {
    for await (const evt of parseSse(fakeStream())) {
        console.log('event:', evt);
    }
})();

// Stage 2 of 2: open the stream, parse, and reconnect on disconnect.
// Resumes via the Last-Event-ID header so the server can replay missed events.

async function* parseSse(chunkIter) {
    let buf = '';
    let event = { type: 'message', data: '', id: '', retry: null };
    for await (const chunk of chunkIter) {
        buf += chunk;
        let nl;
        while ((nl = buf.indexOf('\n')) !== -1) {
            const line = buf.slice(0, nl).replace(/\r$/, '');
            buf = buf.slice(nl + 1);
            if (line === '') {
                if (event.data) yield { ...event, data: event.data.replace(/\n$/, '') };
                event = { type: 'message', data: '', id: event.id, retry: event.retry };
                continue;
            }
            if (line.startsWith(':')) continue;
            const colon = line.indexOf(':');
            const field = colon === -1 ? line : line.slice(0, colon);
            let value = colon === -1 ? '' : line.slice(colon + 1);
            if (value.startsWith(' ')) value = value.slice(1);
            if (field === 'data') event.data += value + '\n';
            else if (field === 'event') event.type = value;
            else if (field === 'id') event.id = value;
            else if (field === 'retry') event.retry = parseInt(value, 10);
        }
    }
}

async function* readChunks(body) {
    // Body is an async iterable of strings or Uint8Arrays.
    const decoder = new TextDecoder();
    for await (const piece of body) {
        yield typeof piece === 'string' ? piece : decoder.decode(piece, { stream: true });
    }
}

async function consumeSse(url, onEvent, { maxBackoffMs = 30_000, signal } = {}) {
    let lastId = '';
    let serverRetryHint = 1000;
    let attempt = 0;
    while (!signal || !signal.aborted) {
        try {
            const headers = { accept: 'text/event-stream' };
            if (lastId) headers['last-event-id'] = lastId;
            const res = await fetch(url, { headers, signal });
            if (!res.ok) throw new Error(`http_${res.status}`);
            attempt = 0; // reset on a successful connect
            for await (const evt of parseSse(readChunks(res.body))) {
                if (evt.id) lastId = evt.id;
                if (evt.retry) serverRetryHint = evt.retry;
                onEvent(evt);
            }
            // Stream closed cleanly; reconnect after the server's hint.
        } catch (err) {
            if (signal && signal.aborted) return;
            // Fall through to reconnect.
        }
        attempt += 1;
        const wait = Math.min(maxBackoffMs, serverRetryHint * Math.min(8, attempt));
        await new Promise((r) => setTimeout(r, wait));
    }
}

// Demo: a fake server that disconnects once, then resumes from Last-Event-ID.
let calls = 0;
globalThis.fetch = async (url, init) => {
    calls += 1;
    const lastId = init.headers && init.headers['last-event-id'];
    if (calls === 1) {
        return {
            ok: true,
            body: (async function* () {
                yield 'id: 1\ndata: hello\n\n';
                yield 'id: 2\ndata: world\n\n';
                // server hangs up
            })(),
        };
    }
    console.log('reconnected with Last-Event-ID:', lastId);
    return { ok: true, body: (async function* () { yield 'id: 3\ndata: resumed\n\n'; })() };
};

const ac = new AbortController();
const events = [];
consumeSse('https://stream.example/feed', (evt) => {
    events.push(evt.data);
    if (events.length === 3) {
        ac.abort();
        console.log('events seen in order:', events);
    }
}, { signal: ac.signal, maxBackoffMs: 50 });