Heute gehen wir anhand von zwei Beispielen auf die Details rund um Streaming Resources sowie auf die semantischen Feinheiten und das Zusammenspiel mit RxJS ein. Den Quellcode finden Sie unter [1].
Beispielanwendungen
Um die Nutzung von Streaming Resources zu demonstrieren, kommen hier zwei Beispiele zum Einsatz. Das erste ist quasi das „Hello World“ der reaktiven Programmierung: ein einfacher Timer, der im Sekundentakt hochzählt (Abb. 1).
Das Schöne an diesem sehr einfachen Beispiel ist, dass es bereits vieles enthält, an dem die kleinen, jedoch wichtigen Feinheiten von Streaming Resources diskutiert werden können. Eine Variante dieses Beispiels kommt auch zum Einsatz, um die Interoperabilität mit RxJS zu zeigen.
Das zweite Beispiel ist ein wenig praxisnäher. Es handelt sich dabei um einen einfachen WebSockets-basierten Chat auf der Basis einer Streaming Resource (Abb. 2).
Dieses Chatbeispiel unterscheidet sich technisch gesehen in einem wichtigen Punkt vom Timer: Hier sind zum selben Zeitpunkt alle empfangenen Nachrichten und nicht nur die letzte von Bedeutung. Das stellt eine kleine Herausforderung in der Signal-basierten Welt dar, die sich allerdings lösen lässt.
Als Chatserver kommt ein Fork eines Beispielprojektes aus dem Mozilla Developer Network zum Einsatz [2]. Dieser Node-basierte Server liegt dem Beispielprojekt [1] bei und die readme.md zeigt, wie er sich starten lässt.
Anatomie einer Streaming Resource
Im Gegensatz zu herkömmlichen Resources liefert der Loader einer Streaming Resource immer die folgende Struktur zurück:
PromiseLike<Signal<
{ value: T; }
| { error: unknown;}>>;
Es handelt sich hierbei um ein Promise mit einem Signal, das ein Objekt vorhält. Das Objekt verweist entweder auf den zuletzt veröffentlichten Wert (value) oder auf einen aufgetretenen Fehler. Dadurch, dass das Signal im Verlauf der Zeit mehrere Werte bzw. Fehlerzustände annimmt, repräsentiert es einen Datenstrom.
Um in weiterer Folge den Programmcode zu vereinfachen, definiere ich für den Inhalt dieses Signals einen Typ namens StreamItem (Listing 1). Angular bietet derzeit dafür keinen expliziten Typ an. Das soll sich aber im Laufe der weiteren Entwicklung noch ändern.
Listing 1
export type StreamItem<T> =
| {
value: T;
}
| {
error: unknown;
};
Um eine Streaming Resource zu definieren, nutzt der Programmcode die Funktion resource, die auch bei herkömmlichen Resources zum Einsatz kommt. Anstatt eines Loaders bekommt diese Funktion jedoch einen Streaming Loader über die Eigenschaft stream übergeben (Listing 2). Da dieses Beispiel das Schlüsselwort async nutzt, ergibt sich das Promise, das das Signal bereitstellt, implizit.
Listing 2
const myResource = resource({
request: requestSignal,
stream: async (params) => {
// 1. Create Signal representing the Stream
const result = signal<StreamItem<number>>({
value: 4711
});
// 2. Set up async logic updating the Signal
[…]
// 3. Set up clean-up handler triggered by AbortSignal
params.abortSignal.addEventListener('abort', () => {
[…]
});
// 4. Return Signal
return result;
},
});
Typischerweise lässt sich ein Streaming Loader in vier Teile untergliedern, die im gezeigten Beispiel mit Kommentaren angedeutet sind:
-
Zunächst erzeugt der Streaming Loader ein neues Signal, das den Datenstrom repräsentiert. Dieses Signal bekommt einen Initialwert.
-
Danach startet der Streaming Loader eine asynchrone Operation, die im Laufe der Zeit mehrere Ergebnisse liefert. Diese Ergebnisse veröffentlicht er nach und nach über das Signal.
-
Der Streaming Loader ist auch für das Bereitstellen einer Aufräumlogik verantwortlich. Diese beendet die zugrundeliegende asynchrone Operation, wenn ihre Werte nicht mehr benötigt werden.
-
Am Ende liefert der Streaming Loader das Signal zurück.
Für das Bereitstellen der Aufräumlogik nutzt der Streaming Loader das vom Resource API bereitgestellte AbortSignal, das sich im übergebenen Parameterobjekt findet. Interessanterweise entsprechen diese Punkte mehr oder weniger auch der typischen Vorgehensweise bei der direkten Nutzung des Observable-Konstruktors in der RxJS-Welt.
switchMap-Semantik beim Übergang zwischen Streams
Die definierte Aufräumlogik läuft immer dann, wenn die Anwendung den aktuellen Datenstrom nicht mehr benötigt. Dafür gibt es zwei Gründe: Der erste tritt dann ein, wenn Angular den Building Block zerstört, der die Resource beherbergt. Stellen wir uns dazu eine Komponente mit einer Resource vor. Beim Verlassen zerstört der Router diese Komponente und im Rahmen dessen zerstört Angular auch die Resource.
Der zweite Grund ergibt sich, wenn sich das request-Signal der Resource ändert. Jede Änderung triggert den Streaming Loader und dieser liefert daraufhin einen neuen Stream. Beim Übergang zwischen diesen Streams nutzt das Resource API somit dieselbe Semantik wie switchMap in RxJS (Abb. 3).
Das bedeutet, dass die Resource immer nur den neuesten Stream konsumiert. Diese Vorgehensweise ist die gleiche, die in der Regel beim Laden von Daten zum Einsatz kommt. Wie generell bei Signals in Angular geht es also auch hier darum, für Standardfälle einfache Konzepte anzubieten. Für komplexere Szenarien kann die Anwendung auf Bibliotheken wie RxJS zurückgreifen.
Ein einfacher Resource-basierter Timer
Nachdem wir nun den prinzipiellen Aufbau einer Streaming Resource besprochen haben, möchte ich die konkrete Nutzung anhand eines ersten Beispiels demonstrieren: Ein Timer, der eine Zahl in einem bestimmten Intervall hochzählt. Listing 3 zeigt diese timerResource aus Sicht eines Konsumenten.
Listing 3
@Component([…])
export class TimerResourceComponent {
ResourceStatus = ResourceStatus;
startValue = signal(0);
timer = timerResource(1000, this.startValue);
forward(): void {
this.startValue.update((v) => nextSegment(v));
}
}
function nextSegment(currentValue: number): number {
return Math.floor(currentValue / 100) * 100 + 100;
}
Der Timer liefert einen Stream, der beim übergebenen Startwert zu zählen beginnt. Der erste Parameter von timerResource repräsentiert das gewünschte Intervall in Millisekunden. Den Startwert repräsentiert das Signal startValue. Immer wenn er sich ändert, wechselt der Timer zu einem neuen Stream. Um diesen Umstand zu demonstrieren, springt die Methode forward zum nächsten vollen Hunderter, also z. B. von 17 auf 100 oder von 123 auf 200.
Factory für die Streaming Resource
Um den Einsatz der Streaming Resource für den Timer zu vereinfachen, handelt es sich bei der Funktion timerResource lediglich um eine Factory (Listing 4).
Listing 4
export function timerResource(
timeout: number,
startValue: () => number
): ResourceRef<number | undefined> {
const request = computed(() => ({
startValue: startValue(),
}));
const result = resource({
request: request,
stream: async (params) => {
const counter = params.request.startValue;
[…]
}
});
return result;
}
Diese Factory nimmt das gewünschte Intervall (timeout) sowie ein Signal mit dem Standardwert entgegen. Bei diesem Signal ist timerResource nur am Getter interessiert. Deswegen typisiert sie das Signal mit () => number. Der Rückgabewert ist eine ResourceRef<number | undefined>. Der Typparameter entspricht den Werten im Stream. Da die Resource den Stream asynchron erzeugen kann, setzt sie initial diesen Wert auf undefined. Um undefined zu vermeiden, wird Angular künftig auch die Möglichkeit bieten, einen eigenen Initialwert festzulegen.
Das Computed Signal request repräsentiert alle Parameter, die den Loader triggern. Da im besprochenen Beispiel lediglich startValue als Trigger zum Einsatz kommt, fühlt sich das Computed Signal ein wenig wie ein Durchlauferhitzer an. Trotzdem halte ich gerne an dieser Vorgehensweise fest, zumal sie es erlaubt, später den Trigger zu erweitern und den Namen startValue auch innerhalb des (Streaming) Loaders verfügbar macht: Dieser kann im diskutierten Fall über param.request.startValue den aktuellen Wert beziehen.
Streaming Loder für den Timer
Der Aufbau des Streaming Loaders entspricht den oben diskutierten vier Abschnitten (Listing 5). Die asynchrone Operation zum Hochzählen implementiert der Loader mit der althergebrachten setInterval-Funktion von JavaScript. Um das Verhalten von Fehlerzuständen zu demonstrieren, meldet der Timer bei den Werten 7 und 13 einen Fehler – abergläubige Benutzer:innen kommen somit auch auf ihre Kosten.
Listing 5
const result = resource({
request: request,
stream: async (params) => {
let counter = params.request.startValue;
// 1. Create Signal representing the Stream
const resultSignal = signal<StreamItem<number>>({
value: params.request.startValue,
});
// 2. Set up async logic updating the Signal
const ref = setInterval(() => {
counter++;
console.log('tick', counter);
if (counter === 7 || counter === 13) {
resultSignal.set({ error: 'bad luck!' });
} else {
resultSignal.set({ value: counter });
}
}, timeout);
// 3. Set up clean-up handler triggered by AbortSignal
params.abortSignal.addEventListener('abort', () => {
console.log('clean up!');
clearInterval(ref);
});
// 4. Return Signal
return resultSignal;
},
});
Durch das asynchrone Verhalten von setInterval findet Schritt 4 vor der ersten Ausführung des Callbacks in Schritt 2 statt. Das bedeutet, dass der Loader zunächst das Signal mit seinem Initialwert zurückliefert und erst dann das Signal nach und nach neue Werte erhält.
Die Streaming Resource ausprobieren
Beim Ausprobieren des Beispiels sieht man zunächst, wie das Beispiel den Zähler im Sekundentakt inkrementiert. Anstatt der Werte 7 und 13 meldet es einen Fehler. Im Gegensatz zu RxJS beendet solch ein Fehler den Stream jedoch nicht. Sobald der Streaming Loader einen neuen Wert liefert, stellt die Resource ihn bereit.
Abbildung 4 zeigt die Konsolenausgaben des Loaders. Sie unterstreichen, dass die Streaming Resource zu einem Zeitpunkt lediglich den aktuellen Stream nutzt und sich somit die switchMap-Semantik ergibt.
Kurz vor jeder Ausgabe von „clean up!“ hat der User die Funktion forward ausgelöst. Dies führt zu einem neuen Wert in startValue und dieser Umstand triggert den Loader erneut. Die Resource benachrichtigt das AbortSignal des alten Streams, den ihr abort-Handler beendet. Fortan nutzt die Resource den vom Loader gelieferten neuen Stream, der in unserem Beispiel mit dem nächsten Hunderter fortsetzt.
Interop mit RxJS und Observables
Gemeinsam mit dem Resource API führte Angular 19 auch die rxResource ein, die den Brückenschlag zur RxJS-Welt erlaubt. Ab Angular 19.2 sind rxResources immer Streaming Resources. Das bedeutet, eine rxResource liefert nach und nach die Werte des vom Loader zurückgelieferten Observables. Um dieses Verhalten zu demonstrieren, zeigt Listing 6 eine Variation des zuvor besprochenen Timers auf Basis von rxResource.
Listing 6
export function timerResource(
timeout: number,
startValue: () => number
): ResourceRef<number | undefined> {
const request = computed(() => ({
startValue: startValue(),
}));
return rxResource({
request: request,
loader: (params) => {
const startValue = params.request.startValue;
return interval(timeout).pipe(
map((v) => v + startValue + 1),
startWith(startValue),
tap((v) => console.log('counter', v)),
switchMap((value) => {
if (value === 7 || value === 13) {
return throwError(() => 'bad luck');
}
return [value];
})
);
},
});
}
Diese Implementierung ist dank der zahlreichen Operatoren, die RxJS anbietet, etwas kürzer. Prinzipiell ließen sich analoge Hilfsfunktionen auch für Signals und Resources bereitstellen. Allerdings habe ich das Gefühl, dass der Trend in der Signals-Welt eher in Richtung anwendungsfallbezogener und somit grobgranularer Building Blocks wie timerResource geht. Die Zeit wird zeigen, ob sich dieser Eindruck bewahrheitet.
Wichtige Details beim Einsatz von Streams mit rxResource
Ein kleines Detail bei der rxResource ist die Tatsache, dass die Anwendung den Streaming Loader nicht über die Eigenschaft stream, sondern über loader bereitstellt. Das mag der Tatsache geschuldet sein, dass ein Loader einer rxResource immer ein Streaming Loader ist. Möchte die Anwendung, wie in der ursprünglichen Implementierung in Angular 19.0 und 19.1 lediglich den ersten Wert des Observables nutzen, muss sie nun auf Operatoren wie first oder take(1) zurückgreifen.
Ein weiterer wichtiger Unterschied ergibt sich beim Auftreten eines Fehlers. Hier prallen die Semantiken von Observables und Resources aufeinander. Ein Observable schließt automatisch beim ersten unbehandelten Fehler. Aus diesem Grund erholt sich das Observable in der rxResource nach dem ersten Fehler, der in unserem Beispiel anstatt des Wertes 7 auftritt, nicht mehr. Allerdings kann sich eine rxResource aus dem Fehlerzustand retten, indem es auf einen neuen Stream wechselt. Dazu muss sich zum Beispiel das request-Signal ändern, das den Loader erneut anstößt, dass dieser ein neues Observable liefert.
Wie gewohnt kommt beim Wechsel auf neue Observables die switchMap-Semantik zum Einsatz. Das heißt, dass auch die rxResource zu einem Zeitpunkt lediglich das neueste Observable nutzt. Vorgänger schließt sie, indem sie sich abmeldet (unsubscribe).
Streaming Resources für einen Chat mit WebSockets
Das bis jetzt betrachtete Minimalbeispiel war bereits umfangreich genug, um die Nutzung von Streaming Resources und ihre semantischen Details zu betrachten. Nun möchte ich jedoch einen Schritt weiter gehen und einen etwas praxisnäheren Anwendungsfall zeigen. Dabei handelt es sich um einen einfachen Chatclient (Abb. 2), der es erlaubt, weitere Details zu diskutieren.
Der Angular-Client initialisiert den Chat mit der Factory chatConnection. Sie liefert eine Struktur, die die einzelnen Chatnachrichten als Streaming Resource repräsentiert und daneben noch weitere Signals sowie eine send-Methode anbietet (Listing 7).
Listing 7
@Component([…])
export class ChatResourceComponent {
ResourceStatus = ResourceStatus;
userName = signal('');
chat = chatConnection('ws://localhost:6502', this.userName);
messages = computed(() => this.chat.resource.value() ?? []);
userNameInField = linkedSignal(() => this.chat.acceptedUserName());
currentMessage = signal<string>('');
send() {
this.chat.send(this.currentMessage());
this.currentMessage.set('');
}
join() {
this.userName.set(this.userNameInField());
}
}
Das an chatConnection übergebene userName-Signal triggert die Resource und führt zum Verbindungsaufbau mit dem Chatserver. Die Signals userNameInField und currentMessages sind an Eingabefelder gebunden. Um eine Verbindung aufzubauen, schreibt die Funktion join den Wert von userNameInField in das Signal userName.
Der Server kann den gewünschten Benutzernamen korrigieren, um für Eindeutigkeit zu sorgen. Deswegen ist userNameInField mit linkedSignal implementiert. Dieses Linked Signal überschreibt den lokalen Wert mit dem korrigierten Wert vom Server, der sich im Signal accpetedUserName befindet.
Möchten Benutzer:innen die aktuelle Nachricht absenden, übergibt die Anwendung die currentMessage an die send-Methode. Daraufhin setzt sie die currentMessage zurück, damit Benutzer:innen die nächste Nachricht erfassen können.
Glitch-free Property als Herausforderung bei Streaming Resources
Das Chatbeispiel unterscheidet sich in einem essenziellen Aspekt vom zuvor besprochenen Timer: Während beim Timer immer nur die neueste Nachricht aus dem Stream von Interesse war, sind es beim Chat sämtliche bisher empfangene Nachrichten oder zumindest alle Nachrichten aus einem definierten Zeitfenster. Anders ausgedrückt repräsentiert der Timer eher ein Objekt, das sich im Laufe der Zeit ändert, während der Chat hingegen mehrere Events mit Nachrichten repräsentiert, die die Anwendung im Laufe der Zeit empfängt und in Ihrer Gesamtheit zu berücksichtigen hat.
Die Art und Weise, wie wir hier den Timer betrachten, passt ganz gut zu Signals, zumal Signals jeweils den neuesten Wert wiederspiegeln. Hier passt auch die Glicht-free-Eigenschaft, die unnötige Zwischenwerte ignoriert: Würde die Resource den Timer in einem Atemzug, also in einem einzigen Task, von 0 auf 1, von 1 auf 2 und von 2 auf 3 ändern, sähe der Konsument der Resource nur den Wert 3.
Genau dieses Verhalten wäre jedoch bei einer Event-basierten Nutzung wie beim Chat fatal. Hier gingen einzelne Nachrichten verloren. Dieses Verhalten lässt sich leicht testen, indem man den Chatserver veranlasst, dieselbe Nachricht hintereinander mehrfach an denselben Client zu senden. Die meisten dieser Wiederholungen würde bei einer Implementierung, die lediglich die zuletzt empfangene Nachricht berücksichtigt, verloren gehen.
Diese Überlegung zeigt, dass Signals im Gegensatz zu Observables in RxJS nicht wirklich gut für das Repräsentieren von Events bzw. Nachrichten geeignet sind. Das stellt uns beim Einsatz von Streaming Resources vor eine Herausforderung, die sich glücklicherweise durch eine einfache Anpassung unserer Sichtweise bewältigen lässt: Lassen wir die Resource nicht nur den aktuellen Wert repräsentieren, sondern alle bisher empfangenen Nachrichten oder zumindest alle Nachrichten eines für uns interessanten Zeitfensters, können wir diese Gesamtheit an Nachrichten wieder als Wert betrachten, der sich im Laufe der Zeit ändert.
Anders ausgedrückt müssen wir das Sammeln und das Verwalten der einzelnen Nachrichten in die Resource verschieben. Diese Sichtweise passt auch gut zu meiner weiter oben erwähnten Beobachtung, wonach in der Signals-Welt eher anwendungsfallspezifische und somit grobgranulare Buildings Blocks zum Einsatz kommen.
Typen für Nachrichten
Für die einzelnen Nachrichten, die der Client mit dem Chatserver austauscht, legt das Beispiel einige Typen fest (Listing 8).
Listing 8
export type ChatRequest =
| {
type: 'username';
id: number;
name: string;
}
| {
type: 'message';
id: number;
text: string;
};
export type ChatResponse =
| {
type: 'id';
id: number;
}
| {
type: 'username';
id: number;
name: string;
}
| {
type: 'message';
id: number;
name: string;
text: string;
};
Das Protokoll zwischen Client und Server gestaltet sich wie folgt:
-
Der Client baut eine WebSockets-Verbindung auf.
-
Der Server reagiert mit einer ChatResponse vom Typ id, über die der Client eine eindeutige Session-ID erhält.
-
Der Client übersendet einen ChatRequest vom Typ username, mit dem er über den Namen des Benutzers informiert.
-
Der Server bestätigt diesen Benutzernamen mit einer ChatResponse vom Typ username. Falls der gewünschte Benutzername schon vergeben ist, erhält der Client über diese Nachricht einen korrigierten Benutzernamen, den der Server durch Anhängen einer Zahl an den Wunschnamen bildet.
-
Der Client übersendet ChatRequests mit seinen Nachrichten (Typ message). Der Server verteilt diese Nachricht in Form einer ChatResponse des Typs message an alle Clients.
Repräsentation des Chats
Die Verbindung mit dem Chat repräsentiert das Beispiel mit dem Typ ChatConnection (Listing 9).
Listing 9
export type SendFn = (message: string) => void;
export type ChatConnection = {
resource: ResourceRef<ChatResponse[] | undefined>;
connected: () => boolean;
acceptedUserName: () => string;
send: SendFn;
};
Das Herzstück der ChatConnection ist eine Resource, die sämtliche bisher empfangene Nachrichten in Form eines Arrays repräsentiert. Daneben liefert sie noch weitere Statusinformationen als Signals. Dazu gehört der Verbindungszustand (connected) und der ggf. vom Server korrigierte Benutzername (accpetedUserName). Die Methode send erlaubt es dem Chat, Nachrichten zum Server zu senden. Den übergebenen String bildet die Implementierung auf einen ChatRequest vom Typ message ab.
Factory für Chat
Die Factory für den Chat ist ähnlich wie die für den zuvor besprochenen Timer aufgebaut. Allerdings bereitet sie anfangs ein paar zusätzliche Variablen vor (Listing 10). Zu diesen Variablen gehört die connection, die die vom Streaming Loader aufgebaute WebSockets-Verbindung widerspiegelt, die zuvor erwähnten Signals connected und accpetedUserName sowie ein Signal id, das die aktuelle Usersitzung repräsentiert. Am Ende liefert die Factory ein paar dieser Signals sowie die erzeugte Streaming Resource und die Methode send als ChatConnection zurück. Das Signal id veröffentlicht sie nicht, da es sich dabei um ein Implementierungsdetail handelt.
Listing 10
export function chatConnection(
websocketUrl: string,
userName: () => string
): ChatConnection {
let connection: WebSocket;
const connected = signal(false);
const id = signal(0);
const acceptedUserName = signal('');
const request = computed(() => ({
userName: userName(),
}));
const chatResource = resource({
request,
stream: async (params) => {
// init web socket connection
// handle and collect messages
[…]
},
});
const send: SendFn = (message: string) => {
const request: ChatRequest = {
type: 'message',
id: id(),
text: message,
};
connection.send(JSON.stringify(request));
};
return {
connected,
resource: chatResource,
acceptedUserName,
send,
};
}
Der Streaming Loader der Resource legt sich zunächst ein Array messages zurecht, in dem er die empfangenen Chatnachrichten sammelt (Listing 11). Die restliche Implementierung folgt den zu Beginn besprochenen vier Punkten. Das vorbereitete Signal spiegelt den Stream wider und nimmt das verwaltete Array mit den empfangenen Nachrichten auf.
Listing 11
const chatResource = resource({
request,
stream: async (params) => {
const userName = params.request?.userName;
let messages: ChatResponse[] = [];
// 1. Create Signal representing the Stream
const resultSignal = signal<StreamItem<ChatResponse>>({
value: messages,
});
if (!userName) {
return resultSignal;
}
// 2. Set up async logic updating the Signal
connection = new WebSocket(websocketUrl, 'json');
connection.addEventListener('open', (event) => {
console.log('[open]');
connected.set(true);
});
connection.addEventListener('message', (event) => {
const value = JSON.parse(event.data) as ChatResponse;
console.log('[message]', value);
if (value.type === 'id') {
id.set(value.id);
sendUserName(value.id, userName, connection);
}
if (value.type === 'username' && value.id == id()) {
acceptedUserName.set(value.name);
}
if (value.type === 'message' || value.type === 'username') {
messages = [...messages, value];
resultSignal.set({ value: messages });
}
});
connection.addEventListener('error', (event) => {
const error = event;
console.log('[error]', error);
resultSignal.set({ error });
});
// 3. Set up clean-up handler triggered by AbortSignal
params.abortSignal.addEventListener('abort', () => {
console.log('clean up!');
connection.close();
connected.set(false);
id.set(0);
acceptedUserName.set('');
});
// 4. Return Signal
return resultSignal;
},
});
function sendUserName(id: number, userName: string, connection: WebSocket) {
const message: ChatRequest = {
type: 'username',
id: id,
name: userName,
};
connection.send(JSON.stringify(message));
}
Über die eingerichteten Event Handler empfängt der Client Informationen vom Server. Das message-Event kümmert sich um das weiter oben beschriebene Protokoll. Es ergänzt das Array messages um alle empfangenen Nachrichten. Das muss immutable, also durch eine flache Kopie, erfolgen, damit das Signal die Änderung wahrnimmt.
Das open-Event setzt connected auf true und das error-Event überführt den aktuellen Fehler in den Stream. Das abort-Event des AbortSignals schließt die WebSockets-Verbindung und setzt die verwalteten Eigenschaften auf ihren Ausgangszustand zurück. Am Ende liefert der Streaming Loader wie gewohnt das Signal zurück, das den Stream widerspiegelt.
Zusammenfassung
Streaming Resources ab Angular 19.2 repräsentieren Datenströme, ohne dabei auf RxJS angewiesen zu sein. Der asynchrone Streaming Loader aktualisiert über die Zeit hinweg ein Signal, das den Stream repräsentiert. Das request-Signal der Resource triggert den Streaming Loader, der einen neuen Datenstrom liefert und den alten beendet – das entspricht der aus RxJS bekannten switchMap-Semantik.
Allerdings repräsentieren Signals in Angular immer nur den aktuellen Wert; Zwischenwerte bei Änderungen, die im selben Task aufeinander folgen, werden ignoriert (Glitch-free-Eigenschaft). Das ist für sich ändernde Zustände, wie einen Zähler, gut geeignet, kann jedoch bei Event-Flüssen – wie einem Chat, der sämtliche empfangene Nachrichten anzeigen soll – zu Datenverlusten führen. Die Lösung liegt darin, die (in einem bestimmten Zeitfenster) empfangenen Nachrichten in der Resource zu sammeln und das so erhaltene Array über die Resource zu veröffentlichen. Dieses Array entspricht somit einem Wert, der sich im Laufe der Zeit ändert.
Die rxResource, mit der sich RxJS integrieren lässt, ist nun standardmäßig eine Streaming Resource. Möchte eine Anwendung lediglich den ersten Wert des zugrundeliegenden Observables nutzen, muss sie Operatoren wie first oder take(1) einsetzen. Im Gegensatz zu Observables wird eine Resource, die auf einen unbehandelten Fehler läuft, nicht geschlossen, sondern kann weiterhin Werte veröffentlichen. Da jedoch die rxResource auf einem Observable basiert, kann sich der aktuelle Stream nicht von einem Fehlerzustand erholen. Allerdings ist es auch hier möglich, auf einen neuen Stream zu wechseln, indem die Anwendung den Streaming Loader erneut anstößt.
Stay tuned
Bei neuen Artikel & Eventupdates informiert werden:
