Streaming Chat in Angular: Reading SSE with fetch, resource stream and Rendering Token by Token with Signals
Since EventSource cannot send a POST body, we read the LLM stream in Angular with fetch, ReadableStream and TextDecoderStream and write the events into signals. In this post we measure a UTF-8 character split at a chunk boundary, a Stop button that reaches the server through AbortController, requestAnimationFrame halving DOM writes, a @for track mistake that recreates the element on every update, and a resource stream that reports resolved while still streaming.
Version note: The code in this post was built and run with Angular 22.2 (TypeScript 6.0, Node.js 24) and .NET 10; browser measurements were taken with Playwright 1.65 in headless Chromium 158, and the model is Ollama 0.34.4 (qwen3:4b).
Guess First
The Bookstore assistant's answer streams at about 40 pieces per second. A fast cloud provider can fit a few hundred pieces into the same time. On every delta event you write the message text into a signal. For an answer of 300 deltas, how many times does Angular write the text to the DOM? And with two bubbles on the screen, how many times is the <article> element of the streaming answer recreated?
One more: the server sends the word Doğu, but the network packet is split right between the two bytes of the letter ğ. What do you see on screen?
In the previous post we set up the architecture of the Bookstore assistant (Bookstore, the online bookstore used as the shared example across this site's series) and the event vocabulary of /chat/sse. In this post we read that stream in Angular: an SSE parser on top of fetch, a store that writes events into signals, a "Stop" button, auto-scroll, and a one-off AI feature with resource. We answer the questions with measurements halfway through.
Why Isn't EventSource Enough?
The browser has a ready-made API for SSE: EventSource. It handles reconnecting, Last-Event-ID and line parsing on its own. The "Server-Sent Events in Depth" post in the Real-time Applications series shows how to wrap it in an Observable in Angular. But the EventSource constructor takes only a URL and a withCredentials option: you cannot pick the method, send a request body, or add headers such as Authorization. A chat request, however, should carry the user's message, later the conversation id and settings, in the body; putting a long message in the query string runs into URL limits and ends up in logs.
So we send the request as a POST with fetch and read the response body as a ReadableStream. In exchange we have to write two things EventSource gives for free: parsing the event stream format, and deciding what happens when the connection drops. The second is the subject of post 3; here we look at the parser.
Reading SSE with fetch
The Node client in the streaming post of the AI with .NET series used a short reader that split on \n\n. While moving it into an Angular service I brought it closer to the rules in the WHATWG HTML standard: a line can end with CRLF, LF or a lone CR, a line starting with a colon is a comment, a single space after the field name is dropped, and several data: lines are joined with line breaks.
export interface SseMessage {
event: string;
data: string;
}
// WHATWG event stream format: a line ends with CRLF, LF or CR; an empty line dispatches the event
export async function* readSse(body: ReadableStream<BufferSource>): AsyncGenerator<SseMessage> {
// TextDecoderStream joins a multi-byte UTF-8 character split at a chunk boundary with the next chunk
const reader = body.pipeThrough(new TextDecoderStream()).getReader();
let buffer = '';
let event = '';
let data: string[] = [];
try {
while (true) {
const { value, done } = await reader.read();
if (done) {
// Per the spec, a last event not closed by an empty line is discarded
return;
}
buffer += value;
// If the \r of a CRLF ended up at the end of a chunk, wait for the \n; otherwise a fake empty line sneaks in
const cut = buffer.endsWith('\r') ? buffer.length - 1 : buffer.length;
const lines = buffer.slice(0, cut).split(/\r\n|\r|\n/);
buffer = lines.pop()! + buffer.slice(cut);
for (const line of lines) {
if (line === '') {
if (data.length > 0) {
yield { event: event || 'message', data: data.join('\n') };
}
event = '';
data = [];
continue;
}
if (line.startsWith(':')) {
continue; // comment line
}
const colon = line.indexOf(':');
const field = colon === -1 ? line : line.slice(0, colon);
let fieldValue = colon === -1 ? '' : line.slice(colon + 1);
if (fieldValue.startsWith(' ')) {
fieldValue = fieldValue.slice(1);
}
if (field === 'event') {
event = fieldValue;
} else if (field === 'data') {
data.push(fieldValue);
}
}
}
} finally {
// If the consumer leaves the loop early (break, abort), close the connection too
await reader.cancel().catch(() => {});
}
}The most important line of the parser is its first one: pipeThrough(new TextDecoderStream()). The network splits the response wherever it likes; a chunk ending in the middle of a character is normal. I tried the second question in Guess First with a small Node script: I split an event containing Doğu into two chunks right after the first byte of ğ (C4 9F in UTF-8), decoded each chunk separately first and then fed them to readSse. The same script also tries a stream that splits a CRLF in half, a comment line, a two-line data and a last event that is not closed by an empty line:
$ node parser-check.ts
chunks: 31 + 6 bytes
naive : "event: delta\ndata: {\"text\":\"Do��u\"}\n\n"
readSse: delta {"text":"Doğu"}
readSse: delta "{\"text\":\"a\"}"
readSse: note "line 1\nline 2"Decoding the chunks one by one with new TextDecoder().decode() replaces ğ with two U+FFFD replacement characters; every answer with Turkish, emoji or Chinese will show broken characters now and then. TextDecoderStream carries the incomplete byte over to the next chunk, so readSse returns the right text. The split CRLF did not produce a fake empty line, the comment line was skipped, and the text "lost" of the last event never arrived: per the spec we discard an event that is not closed by an empty line. That last behavior is deliberate; if the stream was cut off, losing the last event is safer than trying to parse half a JSON.
A TypeScript note: I first typed the parameter as ReadableStream<Uint8Array>, and the TypeScript 6.0 build used by Angular 22.2 stopped with TS2345: Argument of type 'TextDecoderStream' is not assignable to parameter of type 'ReadableWritablePair<string, Uint8Array<ArrayBufferLike>>'. The writable side of TextDecoderStream expects BufferSource; with the parameter typed as ReadableStream<BufferSource>, response.body passes without trouble.
Typed Events: ChatApi
The parser yields raw event and data; toChatEvent from post 1 turns them into a typed ChatEvent. The service that ties them together:
import { Injectable } from '@angular/core';
import { ChatEvent, toChatEvent } from './chat-events';
import { readSse } from './sse';
@Injectable({ providedIn: 'root' })
export class ChatApi {
// EventSource only sends GET and cannot send a body; we use fetch to POST the message
async *stream(message: string, signal: AbortSignal): AsyncGenerator<ChatEvent> {
const response = await fetch('/chat/sse', {
method: 'POST',
headers: { 'Content-Type': 'application/json', Accept: 'text/event-stream' },
body: JSON.stringify({ message }),
signal,
});
if (!response.ok || !response.body) {
throw new Error(`Chat request failed with HTTP ${response.status}`);
}
for await (const { event, data } of readSse(response.body)) {
const chatEvent = toChatEvent(event, data);
if (chatEvent) {
yield chatEvent;
}
}
}
}I chose an AsyncGenerator so the consumer can read with for await, and so that leaving the loop (break, return or an exception) runs the generators' finally blocks as a chain: the finally in readSse closes the connection with reader.cancel(). The response.ok check matters too: SSE events only come with a 200 response; a 400, 429 or 500 comes with a JSON body, and if you hand that to the parser no event comes out and the stream silently ends "empty". We deal with these errors in detail in post 3.
There is a cost: using fetch directly means Angular HttpClient interceptors do not run. If your app adds the token in an interceptor, you need to add the same header here.
Rendering Token by Token with Signals
The store that carries events to the screen keeps the message list in a single signal. In Angular 22 new projects are zoneless and components are OnPush by default; updating a signal is enough to re-render the component that reads it in its template. No async pipe, no markForCheck, no NgZone.run.
import { Injectable, inject, signal } from '@angular/core';
import { ChatApi } from './chat-api';
export type MessageStatus = 'thinking' | 'streaming' | 'done' | 'stopped' | 'interrupted';
export interface ChatMessage {
id: number;
role: 'user' | 'assistant';
text: string;
status: MessageStatus;
}
export interface Usage {
inputTokens: number | null;
outputTokens: number | null;
}
@Injectable({ providedIn: 'root' })
export class ChatStore {
private readonly api = inject(ChatApi);
private nextId = 1;
private controller: AbortController | null = null;
private pendingText = '';
private frame = 0;
readonly messages = signal<ChatMessage[]>([]);
readonly busy = signal(false);
readonly usage = signal<Usage | null>(null);
async send(text: string): Promise<void> {
if (this.busy()) {
return;
}
const userId = this.nextId++;
const id = this.nextId++;
this.messages.update((list) => [
...list,
{ id: userId, role: 'user', text, status: 'done' },
{ id, role: 'assistant', text: '', status: 'thinking' },
]);
this.busy.set(true);
this.controller = new AbortController();
let finished = false;
try {
for await (const event of this.api.stream(text, this.controller.signal)) {
switch (event.type) {
case 'delta':
this.queueDelta(id, event.text);
break;
case 'usage':
this.usage.set({ inputTokens: event.inputTokens, outputTokens: event.outputTokens });
break;
case 'done':
finished = true;
this.flush(id);
this.patch(id, { status: 'done' });
break;
case 'error':
finished = true;
this.flush(id);
this.patch(id, { status: 'interrupted' });
break;
}
}
// A connection that closes without done means "interrupted", not "finished"
if (!finished) {
this.flush(id);
this.patch(id, { status: 'interrupted' });
}
} catch (err) {
this.flush(id);
const stopped = err instanceof DOMException && err.name === 'AbortError';
this.patch(id, { status: stopped ? 'stopped' : 'interrupted' });
} finally {
this.busy.set(false);
this.controller = null;
}
}
stop(): void {
this.controller?.abort();
}
// Collect deltas and write them to the screen once per frame
private queueDelta(id: number, text: string): void {
this.pendingText += text;
if (this.frame === 0) {
this.frame = requestAnimationFrame(() => this.flush(id));
}
}
private flush(id: number): void {
cancelAnimationFrame(this.frame);
this.frame = 0;
if (this.pendingText === '') {
return;
}
const text = this.pendingText;
this.pendingText = '';
this.messages.update((list) =>
list.map((m) => (m.id === id ? { ...m, text: m.text + text, status: 'streaming' } : m)),
);
}
private patch(id: number, change: Partial<ChatMessage>): void {
this.messages.update((list) => list.map((m) => (m.id === id ? { ...m, ...change } : m)));
}
}An assistant message is born in the thinking state, moves to streaming with the first text and ends in one of three states: done (the server sent done), stopped (the user stopped it) or interrupted (the server sent error, the connection closed without done, or a network error happened). The store ignores ping and thinking events; the message already starts in thinking. List updates are immutable: a new object for the message that changed, the same reference for the others. This detail becomes important in a moment with the track expression of @for.
This store deliberately leaves things out: a second send call is silently ignored, error types are not distinguished, and there is no 429 or timeout handling. In post 3 we turn it into an explicit state machine.
The Component: A Stop Button and Auto-Scroll
import { Component, ElementRef, afterRenderEffect, inject, signal, viewChild } from '@angular/core';
import { ChatStore } from './chat-store';
@Component({
selector: 'app-assistant',
templateUrl: './assistant.html',
styleUrl: './assistant.css',
})
export class Assistant {
protected readonly store = inject(ChatStore);
protected readonly draft = signal('');
private readonly log = viewChild.required<ElementRef<HTMLElement>>('log');
private stickToBottom = true;
constructor() {
// Runs after the DOM is updated: scroll to the bottom unless the user scrolled up
afterRenderEffect(() => {
this.store.messages();
const el = this.log().nativeElement;
if (this.stickToBottom) {
el.scrollTop = el.scrollHeight;
}
});
}
protected onScroll(): void {
const el = this.log().nativeElement;
this.stickToBottom = el.scrollHeight - el.scrollTop - el.clientHeight < 40;
}
protected submit(event: Event): void {
event.preventDefault();
const text = this.draft().trim();
if (text === '') {
return;
}
this.draft.set('');
this.stickToBottom = true;
void this.store.send(text);
}
}<section class="log" #log (scroll)="onScroll()">
@for (message of store.messages(); track message.id) {
<article class="bubble" [class.user]="message.role === 'user'">
@if (message.status === 'thinking') {
<p class="status">Thinking…</p>
}
<p class="text">{{ message.text }}</p>
@if (message.status === 'stopped') {
<p class="status">Stopped.</p>
} @else if (message.status === 'interrupted') {
<p class="status error">The answer was interrupted.</p>
}
</article>
} @empty {
<p class="hint">Ask about a book, an order or the store.</p>
}
</section>
<form (submit)="submit($event)">
<input #input [value]="draft()" (input)="draft.set(input.value)" placeholder="Ask the Bookstore assistant" />
@if (store.busy()) {
<button type="button" (click)="store.stop()">Stop</button>
} @else {
<button type="submit" [disabled]="draft().trim() === ''">Send</button>
}
</form>
@if (store.usage(); as usage) {
<p class="usage">Last answer: {{ usage.inputTokens }} input / {{ usage.outputTokens }} output tokens</p>
}The auto-scroll rule is simple: if the user is at the bottom, stay at the bottom as the answer streams; if they scrolled up to read an older message, do not pull them down. afterRenderEffect runs after the DOM is updated, so scrollHeight already includes the new text. While a stream runs, "Stop" replaces "Send"; that shows the way to cancel and also prevents a second message from being sent at the same time at the UI level.
I ran the screen with Playwright against the real model. A MutationObserver recorded the text length and the status line of the last assistant bubble on every change:
$ node run02-chat.mjs full
records: 65 first text at 4851 ms, last at 6302 ms
first 6: [{"t":122,"len":0,"status":"Thinking…"},{"t":4851,"len":3,"status":""},{"t":4885,"len":13,"status":""},
{"t":4902,"len":14,"status":""},{"t":4918,"len":16,"status":""},{"t":4952,"len":19,"status":""}]
answer: For teenagers, *The Hobbit* by J.R.R. Tolkien is a perfect starting point. It offers thrilling adventure, ...
usage : Last answer: 62 input / 272 output tokens
# server log
Stream closed: deltas=64, firstDelta=4744 ms, elapsed=6208 ms, aborted=False"Thinking…" appeared at 122 milliseconds, so for 4.7 seconds the user saw a state instead of an empty screen. The first text arrived at 4851 milliseconds, and the answer streamed in 1.5 seconds. Two things to note: the model wrote three sentences even though I asked for two, and it marked the book title with markdown as *The Hobbit*; on screen the asterisks show up as plain text. Rendering markdown safely is the topic of post 4.
Then I asked for a 300-word story and pressed "Stop" once the text passed 80 characters:
$ node run02-chat.mjs stop
after Stop: "Whiskers, a tabby cat with eyes like old leather, slept on the second-floor shelves of *\n\nStopped."
# server log
Stream closed: deltas=24, firstDelta=15301 ms, elapsed=15836 ms, aborted=Truecontroller.abort() closed the fetch connection, ASP.NET Core carried that through HttpContext.RequestAborted to the endpoint's CancellationToken, and the server closed the stream at the 24th delta (aborted=True). We measured in the streaming post of the AI with .NET series that the token reaches the model and stops Ollama too. The partial text stays on screen with "Stopped." under it; the user knows what they cut off themselves.
requestAnimationFrame and @for track: A Measurement
Now the first question from Guess First. With the real model there are 65 records for 64 deltas: one "Thinking…" and one text change per delta. The deltas arrive about 23 milliseconds apart, which is longer than one frame (16.7 ms at 60 Hz), so batching with requestAnimationFrame gains nothing with this model. To imitate a fast provider I replaced fetch in Playwright with a fake stream that produces 300 deltas, each in its own chunk and 2 milliseconds apart. Across four variants of the store and the template I counted how many times the text was written to the DOM and how many <article> elements were added:
| Variant | Text updates | article elements added | Stream time |
|---|---|---|---|
requestAnimationFrame + track message.id | 91 | 2 | 1542 ms |
Write on every delta + track message.id | 183 | 2 | 1570 ms |
requestAnimationFrame + track message | 90 | 91 | 1551 ms |
Write on every delta + track message | 184 | 185 | 1583 ms |
Three results. First, even without requestAnimationFrame 300 deltas did not become 300 renders but 183. Zoneless Angular merges signal changes made in the same task into one change detection; when the browser delivers several events in the same chunk, they are written in one go anyway. Second, requestAnimationFrame cut the writes down to the screen refresh rate, about 90 in 1.5 seconds; there is no visible difference, but half the work. That gap matters much more in post 4, when we start parsing markdown on every update. Third and most important is track. track message tracks by object identity; with immutable updates the streaming message is a new object every time, so Angular removed and recreated its <article> on every update: 91 elements in 1.5 seconds. Text the user selected, an open menu, focus in an input are lost on every update. With track message.id only two <article> elements were added (user and assistant), and the text of the same element was updated throughout the stream.
A One-Off AI Feature with resource stream
In post 1 we said an AI interface is not just a chat box. Showing a "why you should read this book" sentence as streaming text on the book detail page is one request with one value; it needs no message list. The stream option of resource, stable in Angular 22, is made for this: the loader returns a Signal<ResourceStreamItem<T>>, and each new piece is written to that signal as { value } or { error }.
import { Component, ResourceStreamItem, inject, input, resource, signal } from '@angular/core';
import { ChatApi } from '../assistant/chat-api';
export interface Pitch {
text: string;
done: boolean;
}
@Component({
selector: 'app-book-pitch',
template: `
@if (pitch.error(); as error) {
<p class="error">No summary right now: {{ error.message }}</p>
} @else if (pitch.hasValue()) {
<p>{{ pitch.value().text }}@if (!pitch.value().done) {<span class="cursor">▍</span>}</p>
} @else {
<p class="muted">Generating a summary…</p>
}
`,
})
export class BookPitch {
readonly title = input.required<string>();
private readonly api = inject(ChatApi);
// One request, one value: when the book changes, the previous stream is cancelled through abortSignal
protected readonly pitch = resource({
params: () => ({ title: this.title() }),
stream: async ({ params, abortSignal }) => {
const result = signal<ResourceStreamItem<Pitch>>({ value: { text: '', done: false } });
const prompt = `In one sentence, why should someone read "${params.title}"?`;
void (async () => {
let text = '';
try {
for await (const event of this.api.stream(prompt, abortSignal)) {
if (event.type === 'delta') {
text += event.text;
result.set({ value: { text, done: false } });
} else if (event.type === 'done') {
result.set({ value: { text, done: true } });
} else if (event.type === 'error') {
result.set({ error: new Error(event.message) });
}
}
} catch (err) {
// Don't show an error if it was cancelled because the book changed
if (!abortSignal.aborted) {
result.set({ error: err instanceof Error ? err : new Error(String(err)) });
}
}
})();
return result;
},
});
}I put the done field into the value type on purpose. The reason is what I saw when I sampled the resource status throughout the stream with ng.getComponent in Playwright:
{"t":58,"status":"resolved","isLoading":false,"hasValue":true,"text":0,"done":false}
{"t":5907,"status":"resolved","isLoading":false,"hasValue":true,"text":2,"done":false}
{"t":6535,"status":"resolved","isLoading":false,"hasValue":true,"text":169,"done":true}
text: "Dune" offers a groundbreaking blend of political strategy, ecological philosophy, and intricate world-building ...The moment the loader returned the signal, at 58 milliseconds and before a single word arrived, the resource became resolved and isLoading went back to false. The resource does not know when the stream ends; as far as it is concerned, the value is "resolved" and then changes. If you tie a loading indicator to isLoading(), it disappears while the model is still thinking. You need to carry the end of the stream inside the value; here the done field and the cursor (▍) do that.
To see abortSignal at work I switched books while the stream was running:
$ node run02-pitch.mjs switch
{"t":7349,"status":"resolved","isLoading":false,"hasValue":true,"text":5,"done":false}
{"t":7503,"clicked":"Emma"}
{"t":7804,"status":"resolved","isLoading":false,"hasValue":true,"text":0,"done":false}
# server log
Stream closed: deltas=11, firstDelta=7307 ms, elapsed=7533 ms, aborted=TrueWhen the route parameter changed, params produced a new value, Angular cancelled the previous loader's abortSignal, fetch closed and the server dropped the Dune stream at the 11th delta. There is no unsubscribe, takeUntil or ngOnDestroy in the code; the same signal is cancelled when the component is destroyed. I did not take this route for the chat because of its shape: a chat is not one value but a growing list of messages, user actions (stop, regenerate) and a state per message. resource is the right tool for work that fits the "reload when the parameter changes" model.
Production Notes
One connection per stream. Over HTTP/1.1 a browser opens at most six connections to the same origin; every open stream holds one of them. If the user has the assistant open in three tabs, other API calls can end up waiting. HTTP/2 or HTTP/3 in production removes this limit; still, cancelling the stream when the user leaves the page is a good habit.
Rendering cost grows with the answer. Here every update changes a single text node and is cheap. Once you add markdown, code highlighting or citations, the work per update grows; repeat the measurement then.
Do not show error messages to the user as they are. TypeError: Failed to fetch thrown by fetch or the server's internal error message tells the user nothing. Here we only showed "interrupted"; what to say in which case comes in post 3.
Focus and screen readers. Text that changes on every token is a separate problem for screen readers; we add aria-live and focus management in post 7.
Common Pitfalls
First, decoding chunks one by one with TextDecoder.decode(). English answers look fine in tests; the first time a Turkish character falls on a chunk boundary, �� appears on screen. Use TextDecoderStream or decode(chunk, { stream: true }).
Second, tracking by object identity in @for. Combined with immutable updates, the DOM of the streaming message is recreated on every update; 91 times in 1.5 seconds in the measurement.
Third, reading the body as SSE without checking response.ok. No event comes out of the JSON body of a 429 or 500 response, and the UI thinks "an empty answer arrived".
Fourth, using isLoading() of a resource stream to mean the stream is still running. Once the loader returns the signal the status is resolved; carry the end inside the value.
Fifth, implementing "Stop" only in the UI: leaving the loop with break without cancelling fetch. The text on screen stops, but the connection and the model on the server keep running. Passing the AbortController signal to fetch is what produces the server's aborted=True log line.
A Small Experiment
★☆☆ Replace requestAnimationFrame in the store with a flush on every delta and ask the real model for an answer. Does the number of text updates change? Then repeat it with the fast fake stream and compare the two results.
★★☆ Add the id: field to readSse and return the last event id with every message. Verify it arrives end to end by giving SseItem an EventId on the server. How would you use this id when reconnecting?
★★★ Write BookPitch with rxResource and an Observable instead of resource: turn ChatApi.stream into an Observable, and see in the server log that the previous subscription is cancelled when the book changes. Which version has less code and less room for mistakes?
The screen now has a streaming answer and a working "Stop". But the store is still optimistic: what happens if the connection never opens, the first token does not arrive for 60 seconds, the server returns 429, or the stream drops halfway and the user wants to regenerate the same question, is undefined. In the next post we turn the chat into an explicit state machine and decide what to show the user for each kind of error.