Auf dieser Seite
RxJS Serie - Observables
Ammar Najjar · Veröffentlicht am 1. Januar 2019RxJS Serie
RxJS (auf Github) ist eine Bibliothek für reaktive Programmierung in Javascript, und in dieser Serie möchte ich meinen Weg des Lernens mit euch teilen.
Aus Wikipedia:
In der Informatik ist reaktive Programmierung ein deklaratives Programmierparadigma, das sich mit Datenströmen und der Propagierung von Änderungen befasst. Mit diesem Paradigma ist es möglich, statische (z. B. Arrays) oder dynamische (z. B. Ereignisemitter) Datenströme einfach auszudrücken und auch mitzuteilen, dass eine abgeleitete Abhängigkeit im zugehörigen Ausführungsmodell existiert, was die automatische Propagierung des geänderten Datenflusses erleichtert.
Aus der Angular-Dokumentation:
Reaktive Programmierung ist ein asynchrones Programmierparadigma, das sich mit Datenströmen und der Propagierung von Änderungen befasst. RxJS (Reactive Extensions for JavaScript) ist eine Bibliothek für reaktive Programmierung mit Observables, die das Komponieren asynchroner oder Callback-basierter Code erleichtert.
Ich werde zunächst die grundlegenden Bausteine und dann die am häufigsten verwendeten Operatoren behandeln.
Bausteine
Observable & Observer & Subscriber
Die Konzepte von Observern und Observables sind miteinander verbunden, denn ein funktionsfähiges Observable gibt es nicht ohne einen Observer.
Schauen wir uns zuerst die Dokumentation an:
Observer ist (aus der Dokumentation):
Eine Schnittstelle für einen Konsumenten von Push-basierten Benachrichtigungen, die von einem Observable geliefert werden.
Subscriber ist (aus der Dokumentation):
Implementiert die Observer-Schnittstelle und erweitert die Subscription-Klasse. Während der Observer die öffentliche API zum Konsumieren der Werte eines Observables ist, werden alle Observer in einen Subscriber umgewandelt, um capabilities wie unsubscribe bereitzustellen.
Observable ist (aus der Dokumentation):
Eine Darstellung einer beliebigen Menge von Werten über eine beliebige Zeitspanne. Dies ist der grundlegendste Baustein von RxJS.
Der Subscriber ist also ein Observer, und alle Observer werden in Subscriber umgewandelt, um die Abonnieren/Deabonnieren-Funktionalität zu unterstützen.
Von Grund auf
Der Quellcode eines Observers sieht so aus:
interface Observer<T> {
closed?: boolean;
next: (value: t) => void;
error: (err: any) => void;
complete: () => void;
}Er implementiert drei Methoden:
next: ein Callback zum Empfangen von Benachrichtigungen des Typsnexterror: ein Callback zum Empfangen von Benachrichtigungen des Typserrorcomplete: ein Callback zum Empfangen von Benachrichtigungen des Typscomplete
Das Observable kann dem Observer Benachrichtigungen pushen, wann immer es möchte, d. h. man könnte eine Funktion implementieren, das das tut:
function customObservable(observer: Observer<T>){
observer.next(1);
observer.next(2);
observer.complete();
}Wenn wir also unser customObservable mit einem Observer als Parameter aufrufen:
customObservable({
next: val => console.log(val),
error: err => console.error(err),
complete: () => console.log('Completed')
});Die Ausgabe sieht so aus:
1
2
CompletedDies ist eine grundlegende Form eines Observables, aber wenn es so einfach wäre, warum bräuchte man ein ganzes Framework dafür? Die Antwort ist einfach: Versucht einmal, next nach complete aufzurufen:
function customObservable(observer: Observer<T>){
observer.next(1);
observer.next(2);
observer.complete();
observer.next(3);
}Die Ausgabe sieht so aus:
1
2
Completed
3Das funktioniert so nicht richtig, wie gewünscht.
Nehmen wir ein komplexeres Beispiel, bei dem das Observable einen Zahlenstrom über ein Intervall einpackt:
function customObservable(observer: Observer<T>){
let counter = 1;
const out = setInterval(() => observer.next(counter++), 1000);
return () => {
console.log('done');
clearInterval(out);
};
}
const streamObservale = customObservable({
next: val => console.log(val),
error: err => console.error(err),
complete: () => console.log('Completed')
});
setTimeout(() => {
streamObservale();
}, 3100);Die Ausgabe erscheint ein Wert pro Sekunde:
1
2
3
doneEin Observable ist also im Grunde ein Wrapper über einen Strom oder eine Menge von Werten, der sie in der erforderlichen Form darstellt. Ein Strom ist eine Sequenz von laufenden Ereignissen, die chronologisch geordnet sind.
Die Beziehung zwischen einem Observer und einem Observable wird in der folgenden Animation gezeigt:
Das Observable kann entweder einen Wert, einen Fehler oder ein Complete-Benachrichtigungssignal ausgeben, das angibt, dass der Strom beendet ist. Der Observer erfasst diese ausgegebenen Ereignisse durch Definition von Funktionen, die auf jedes Ereignis reagieren. Die Art und Weise, wie ein Observer ein Observable überwacht, wird “Subscription” genannt.
Durch die Definition eines Observers wird der Strom definiert, und durch das Abonnieren werden Aktionen für die ausgegebenen Ereignisse definiert. Logisch gesehen ist für jedes Observable eine Subscription erforderlich, um Aktionen zu erstellen, und das wird in den folgenden Beispielen deutlich.
Solange man mehr als eine Menge oder einen Strom von Werten hat, kann man sich vorstellen, wie viele Operationen man darauf anwenden müsste. RxJS bietet viele Möglichkeiten, ein Observable mit allen Sicherheits- und Asynchronitätsfunktionen zu erstellen, die man sich wünschen könnte.
Observable mit RxJS-Statischen Methoden erstellen
- of:
Durch Verpacken einer Wertesequenz mit of wird ein Observable mit diesen Werten zurückgegeben, das die Werte in der gegebenen Reihenfolge ausgibt und dann abschließt.
import { of } from 'rxjs';
const observableOf = of(1, 2, 3); // an observable of the values 1,2,3
observableOf.subscribe(value => console.log(value));Nach oben (Auf StackBlitz ausprobieren)
- range:
Das erinnert mich an Pythons range-Funktion, sie gibt eine Zahlensequenz aus, die durch die Bereiche-Parameter (start, count) bestimmt wird.
import { range } from 'rxjs';
const source = range(10, 5); // 10, 11, 12, 13, 14
source.subscribe(x => console.log(x));Nach oben (Auf StackBlitz ausprobieren)
- interval:
Dies nimmt einen Parameter als Zeit in Millisekunden und erzeugt jede Zeitspanne eine Zahlensequenz. Auf diese Weise erhält man eine unendliche Zeitreihe von Zahlen.
import { interval } from 'rxjs';
const source = interval(1000); // every 1 second
source.subscribe(x => console.log(x));Nach oben (Auf StackBlitz ausprobieren)
- timer:
Dies nimmt zwei Parameter (initialDelay, period). Die anfängliche Verzögerung kann eine Zahl sein, die die Zeit in Millisekunden repräsentiert, oder ein Date-Objekt. Das Observable wartet auf den Wert der anfänglichen Verzögerung und beginnt dann, Zahlen ab 0 in jedem Zeitraum auszugeben.
import { timer } from 'rxjs';
const source = timer(2000, 3000); // start after 2 second, then every 3 seconds
source.subscribe(x => console.log(x));Nach oben (Auf StackBlitz ausprobieren)
- from:
Dies ist eine magische Methode, die so viele Datentypen in Observables umwandelt: Array, Array-ähnlich, Promise, iterierbares Objekt, String (als Array von Zeichen).
import { from } from 'rxjs';
import { take } from 'rxjs/operators';
const fromArray = from([1, 2, 3]); // observable of the values 1,2,3
fromArray.subscribe(value => console.log(value));
// infinite iterable of values
function* generateNumbers(seed) {
let i = seed;
while(true) {
yield i;
i *= 2;
}
}
const iter = generateNumbers(100);
const fromGenerator = from(iter).pipe(take(5)); // 100, 200, 400, 800, 1600
fromGenerator.subscribe(value => console.log(value));
// from a promise
const promise = new Promise((resolve, reject) => {
resolve(true);
reject(false);
});
const source = from(promise);
source.subscribe(x => console.log(x));Nach oben
(Auf StackBlitz ausprobieren) >> take-Operator wird hier verwendet, um die ersten 5 Werte des unendlichen Stroms anzuzeigen, er wird später ausführlich besprochen
- fromEvent:
Diese Methode ist nützlich bei der Arbeit mit dem DOM, da sie Observables aus DOM-Ereignissen oder Node.js-EventEmitter-Ereignissen erstellt.
Sie nimmt zwei Parameter (element: EventTarget, eventName: string)
import { fromEvent } from 'rxjs';
const source = fromEvent(document, 'click'); // capture click events
// show mouse coordinates
source.subscribe(point => console.log(`(${point.clientX}, ${point.clientY})`));Nach oben (Auf StackBlitz ausprobieren)
- create:
Dies erstellt ein benutzerdefiniertes Observable, das die angegebene Funktion ausführt, wenn ein Observer es abonniert. Es kann auch eine Funktion zurückgeben, die ausgeführt wird, wenn der Observer das Abonnement beendet. Dies ist die flexibelste Methode zur Erstellung eines Observables.
Im folgenden Beispiel wird ein Zähler eingerichtet, der einmal pro Sekunde zählt, und dieser Zähler ist ein Observable, das die Funktion zurückgibt, die ausgeführt wird, wenn der Observer das Abonnement beendet.
import { Observable } from 'rxjs';
const source = Observable.create((observer) => {
let id = 0; // setting up a counter
setInterval(() => {
observer.next(id++);
}, 1000);
return () => { id = 0; console.log('Counter cleared');}
})
const sub = source.subscribe(val => console.log(val));
// unsubscribe after 4.5 sec
setTimeout(() => {
sub.unsubscribe();
}, 4500);Nach oben (Auf StackBlitz ausprobieren)
Fazit
Jetzt sollten die gängigsten Möglichkeiten zur Erstellung eines Observables in RxJS klar sein, und ich kann mit dem nächsten Schritt bei den Bausteinen von RxJS fortfahren, nämlich den am häufigsten verwendeten Operatoren für Observables.
Bis zum nächsten Blogbeitrag!
Weitere Referenzen zu Observables: