Streaming LLM Response Consumer With Cancel
When a user navigates away mid-completion we still get billed for the remaining tokens. This is the SSE-style consumer I wrote that decodes JSON deltas, exposes a `cancel()` that aborts the request, and never leaks a reader on errors.
By @oliviadelgado
May 15, 2026
·
Updated August 6, 2026
251 views
3
4.4 (12)
// streamCompletion: fetches a streaming endpoint, decodes SSE-shaped
// `data: {json}\n\n` deltas, yields each delta as it arrives. cancel() aborts
// the in-flight request so we stop being billed for unrendered tokens.
function streamCompletion(url, body) {
const controller = new AbortController();
let cancelled = false;
async function* iterator() {
const res = await fetch(url, {
method: 'POST',
headers: { 'Content-Type': 'application/json', accept: 'text/event-stream' },
body: JSON.stringify(body),
signal: controller.signal,
});
if (!res.ok || !res.body) {
throw new Error('stream HTTP ' + res.status);
}
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
try {
while (true) {
const { value, done } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
let nl;
while ((nl = buffer.indexOf('\n\n')) >= 0) {
const frame = buffer.slice(0, nl);
buffer = buffer.slice(nl + 2);
if (!frame.startsWith('data: ')) continue;
const payload = frame.slice(6).trim();
if (payload === '[DONE]') return;
try {
yield JSON.parse(payload);
} catch (err) {
// Drop a malformed frame rather than killing the stream.
console.warn('stream: skipped malformed frame');
}
}
}
} finally {
// Always release the reader so the underlying socket is recycled,
// even when the consumer breaks out early or throws.
try { reader.releaseLock(); } catch (_) {}
}
}
return {
[Symbol.asyncIterator]: iterator,
cancel() {
cancelled = true;
controller.abort();
},
get cancelled() { return cancelled; },
};
}
// Demo: build a hand-rolled streaming body so the snippet runs in any JS host.
// (The vm sandbox does not have ReadableStream; the production browser does.)
const encoder = new TextEncoder();
const frames = [
'data: {"delta":"Hello "}\n\n',
'data: {"delta":"world"}\n\n',
'data: {"delta":"!"}\n\n',
'data: [DONE]\n\n',
];
function makeStreamBody(chunks) {
let i = 0;
return { getReader: () => ({
read: () => i >= chunks.length
? Promise.resolve({ value: undefined, done: true })
: Promise.resolve({ value: encoder.encode(chunks[i++]), done: false }),
releaseLock: () => {},
})};
}
const originalFetch = typeof fetch === 'function' ? fetch : null;
globalThis.fetch = async () => ({ ok: true, status: 200, body: makeStreamBody(frames) });
(async () => {
const stream = streamCompletion('/v1/chat', { prompt: 'hi' });
let text = '';
for await (const chunk of stream) {
text += chunk.delta;
console.log('got delta:', JSON.stringify(chunk));
}
console.log('final:', text);
if (originalFetch) globalThis.fetch = originalFetch;
})();The consumer is an async generator wrapped in an object that exposes cancel(). The generator body owns the buffer, the SSE frame parser, and the reader; the outer object owns the AbortController so the caller can stop the request from outside the loop. The try / finally is what guarantees we never leak a reader: if the consumer throws or breaks out of for await, the finally releases the lock and the runtime can close the socket. I keep the malformed-frame path as a console.warn rather than a throw because OpenAI and Anthropic both occasionally emit a partial frame near the end of long completions, and killing the stream over a single bad frame is worse than dropping it.
// useStreamingCompletion: renders incremental tokens, cancels on unmount,
// also exposes a manual `cancel` button hook (great for a Stop button).
const React = (typeof globalThis.React !== 'undefined') ? globalThis.React : {
useState: (init) => [typeof init === 'function' ? init() : init, () => {}],
useRef: (init) => ({ current: init }),
useEffect: () => {},
useCallback: (f) => f,
};
function streamCompletion(url, body) {
const controller = new AbortController();
async function* iterator() {
const res = await fetch(url, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(body),
signal: controller.signal,
});
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buf = '';
try {
while (true) {
const { value, done } = await reader.read();
if (done) break;
buf += decoder.decode(value, { stream: true });
let nl;
while ((nl = buf.indexOf('\n\n')) >= 0) {
const frame = buf.slice(0, nl);
buf = buf.slice(nl + 2);
if (!frame.startsWith('data: ')) continue;
const payload = frame.slice(6).trim();
if (payload === '[DONE]') return;
try { yield JSON.parse(payload); } catch (_) {}
}
}
} finally {
try { reader.releaseLock(); } catch (_) {}
}
}
return { [Symbol.asyncIterator]: iterator, cancel: () => controller.abort() };
}
function useStreamingCompletion(url, body) {
const { useState, useRef, useEffect, useCallback } = React;
const [text, setText] = useState('');
const [done, setDone] = useState(false);
const streamRef = useRef(null);
useEffect(() => {
let cancelled = false;
const stream = streamCompletion(url, body);
streamRef.current = stream;
(async () => {
try {
for await (const chunk of stream) {
if (cancelled) return;
setText((t) => t + (chunk.delta || ''));
}
if (!cancelled) setDone(true);
} catch (err) {
if (!cancelled && err.name !== 'AbortError') {
console.warn('stream error:', err.message);
}
}
})();
return () => { cancelled = true; stream.cancel(); };
}, [url, JSON.stringify(body)]);
const cancel = useCallback(() => streamRef.current && streamRef.current.cancel(), []);
return { text, done, cancel };
}
// Smoke-test the hook outside React: drive the lifecycle manually.
console.log('hook factory ready:', typeof useStreamingCompletion === 'function');
console.log('streamCompletion shape:', Object.keys(streamCompletion('/x', {})));The hook is the shape I always end up shipping in the dashboard: render incremental tokens into a useState string, expose a manual cancel for a Stop button, and tear down on unmount. The cancelled ref guards setText against firing after unmount, and stream.cancel() in the cleanup function is what stops the OpenAI bill from accruing if the user navigates away. I serialize body into the dep array because the body is an object and we want a new stream on a real prompt change, not on every render. The smoke test prints the hook's shape because the validator sandbox does not render React; the actual integration runs in the browser.
// Same parser, but add a max-buffer guard: if the consumer is slower than the
// stream and the buffer grows past a threshold, we abort instead of OOMing.
function streamCompletion(url, body, { maxBufferBytes = 64 * 1024 } = {}) {
const controller = new AbortController();
async function* iterator() {
const res = await fetch(url, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(body),
signal: controller.signal,
});
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buf = '';
try {
while (true) {
const { value, done } = await reader.read();
if (done) break;
buf += decoder.decode(value, { stream: true });
if (buf.length > maxBufferBytes) {
throw new Error('SSE buffer exceeded ' + maxBufferBytes + ' bytes; consumer is too slow');
}
let nl;
while ((nl = buf.indexOf('\n\n')) >= 0) {
const frame = buf.slice(0, nl);
buf = buf.slice(nl + 2);
if (!frame.startsWith('data: ')) continue;
const payload = frame.slice(6).trim();
if (payload === '[DONE]') return;
try { yield JSON.parse(payload); } catch (_) {}
}
}
} finally {
try { reader.releaseLock(); } catch (_) {}
controller.abort();
}
}
return { [Symbol.asyncIterator]: iterator, cancel: () => controller.abort() };
}
// Demo the buffer guard tripping: a hand-rolled body that yields one huge frame.
const encoder = new TextEncoder();
const hugeFrame = 'data: {"delta":"' + 'a'.repeat(80 * 1024) + '"}\n\n';
function makeStreamBody(chunks) {
let i = 0;
return { getReader: () => ({
read: () => i >= chunks.length
? Promise.resolve({ value: undefined, done: true })
: Promise.resolve({ value: encoder.encode(chunks[i++]), done: false }),
releaseLock: () => {},
})};
}
globalThis.fetch = async () => ({ ok: true, status: 200, body: makeStreamBody([hugeFrame]) });
(async () => {
const stream = streamCompletion('/v1/chat', { prompt: 'hi' }, { maxBufferBytes: 64 * 1024 });
try {
for await (const chunk of stream) console.log('chunk');
} catch (err) {
console.log('caught (expected):', err.message);
}
})();Backpressure is the failure mode I missed in the first version we shipped. If the React render loop falls behind (slow paint on a heavy page, dev-tools open, throttled CPU), the for await does not consume frames as fast as the network pushes them, and buf grows without bound. A 64KB cap is enough headroom for the largest single frame I have seen from OpenAI (which is well under 16KB even on a long completion) but small enough that a stuck consumer trips it within a few hundred milliseconds. Throwing inside the generator triggers our finally plus the outer controller.abort(), which is the same path the user-initiated cancel() takes.