Angular(2+) und der Einstieg in die Welt von RxJS

Vorteile von Observables

Observables können Promises vollständig ersetzen. Dies ist jedoch nur ein kleiner Bestandteil des RxJS Frameworks. Die wirklichen Stärken von RxJS kommen erst zum Vorschein, wenn Observables nicht als einmalig ausgeführte Konstrukte, sondern als Stream behandelt werden. Ein Beobachter abonniert ein Observable und kann so über einen längeren Zeitraum auf Ereignisse reagieren.

Beispiel AlertComponent

Als eine der ersten Dateien in unserem neuen Angular5 Projekt habe ich den AlertService erstellt. Dieser beinhaltet ein Subject (Ein Subject ist ein Observable, auf dessen Methoden auch von außen zugegriffen werden kann) und je eine Methode zum Anzeigen einer Erfolgsmeldung sowie einer Fehlermeldung. Eine Angular-Komponente, die AlertComponent, bekommt diesen AlertService über Constructor Injection übergeben und abonniert anschließend das beschriebene Subject und fügt die erhaltenen Alert-Objekte in ein Array des ViewModels ein.

Möchte ich nun innerhalb der Anwendung einen Alert erzeugen, muss ich nur die Methode des AlertService aufrufen und der Alert erscheint auf dem Bildschirm, da die AlertComponent über den Wunsch zur Alert-Erzeugung über das Subject erfährt. Erzeuge ich keinen Alert, tut die Komponente nichts. Sie reagiert also nur dann, wenn tatsächlich etwas getan werden muss. Dies spart Ressourcen.

In AnglarJs hätte man diesen Anwendungsfall möglicherweise unter zu Hilfenahme eines Watchers gelöst, der ständig ein Array oder Objekt auf Veränderung überprüft. Da dies alle paar Millisekunden geschieht, ist dies ein nicht zu verachtender Rechenaufwand, der somit vermieden werden kann.

Die wichtigsten Operatoren im Detail

RxJS stellt zur Erstellung/Bearbeitung/Aggregierung dieser Streams eine Reihe von Operatoren zur Verfügung. Einige der nachfolgenden Operatoren möchte ich im Anschluss kurz vorstellen: 

• Aggregieren von verschiedenen Observables, zum Beispiel durch folgende Operatoren
  • concat
  • zip
  • combineLatest
• Manipulieren, der von Observables gelieferten Werte durch die Operatoren
  • map
  • concatMap, mergeMap(flatMap, selectMany), switchMap
  • reduce, scan
• Realisieren von Intervallen und Timern mit
  • Observable.timer(…)
  • Observable.interval(…)
• Filtern von Mengen mit
  • first, last, filter
• Gruppieren, Transformieren und Zerteilen von Mengen mit
  • groupBy, pluck und partition
• transparentes, nebenläufiges Ausführen von Code mit der .do Funktion

Grundsätzlich gilt: Ein definiertes Observable oder eine Kette von Observables wird erst dann ausgeführt, wenn es einen Beobachter gibt. Dies ist dann der Fall, wenn mindestens an einer Stelle die .subscribe Funktion des Observables ausgeführt wird und der Stream somit geöffnet ist. An dieser Stelle sei gesagt, dass es neben den normalen Observables und Subjects noch andere, leicht abgewandelte Observables, wie zum Beispiel das BehaviorSubject gibt. Auf diese speziellen Typen werde ich in diesem Rahmen jedoch nicht eingehen können. Das Verhalten der nachfolgend vorgestellten Operatoren kann sich daher mit den verschiedenen Variationen von Observables ggf. leicht unterscheiden.

Mithilfe der oben aufgelisteten Operatoren lassen sich die Ausgaben von Observables verändern, zusammenfassen oder auch filtern. Die Liste von Operatoren ist nicht vollständig. Eine vollständige Liste aller RxJS Funktionen kann der offiziellen Dokumentation entnommen werden.

Da RxJS inzwischen bei Version 5.5 angekommen ist, und in der vergangenen Zeit einige Refactorings erfahren hat, haben einige Operatoren mehrere Namen (Aliases). Dies geschieht einerseits, um einen Wiedererkennungswert zwischen den verschiedenen ReactiveX-Versionen zu schaffen, anderseits aber auch, um auf die programmiersprachen-spezifischen Funktionsnamen einzugehen. In der Programmiersprache Scala beispielsweise gibt es die Funktion foldLeft von Haus aus für Arrays. Darum hat RxScala die den Funktionsnamen foldLeft als Alias für reduce adaptiert.

Do

Mit dem Do-Operator können Aktionen ausgeführt werden, ohne dass diese das Observable manipulieren. Im Grunde ist der Do-Operator einfach ein zwischengeschaltetes Observable, welches das vorher definierte abonniert, und selber die Werte mithilfe der next Funktion weiterreicht. Der Do-Operator wird ausgeführt kurz bevor das Event beim Beobachter ankommt. Dieser Operator wird häufig für Log-Ausgaben verwendet.

Map

Mithilfe des Map-Operators können die vom Observable ausgegebenen Ereignissee manipuliert werden, bevor sie beim Beobachter ankommen. Map erwartet eine Funktion, die den Datentyp des Observables zurückgibt. Hier ein Beispiel: Das Quell-Observable gibt „this is a“ aus. Die Map-Funktion manipuliert das Ereignis so, dass es zusätzlich das Wort „string“ anhängt. Beim Beobachter des Observables kommt somit „this is a string“ an.

flatMap

flatMap sowie dessen Aliases mergeMap und selectMany dient wie map ebenfalls dazu, ein Observable zu manipulieren, akzeptiert jedoch eine Funktion, die ein Observable zurückliefert. FlatMap abonniert dann dieses Observable aus dem flatMap Block (auch inner Observable genannt) und liefert dem Beobachter dessen Ergebnis.

Hier das obige Beispiel noch einmal, nur mit FlatMap anstelle von Map.

Würde anstelle von Observable.of die Anweisung Observable.from([…]) verwendet, also würde ein Observable von einem Array erstellt werden und somit mehrere Ereignisse ausgeben, würde flatMap diese parallel abonnieren und alle ausgegebenen Ereignisse an die Beobachter weiterreichen. Der FlatMap-Operator nimmt als zweites Argument eine Zahl entgegen. Diese bestimmt, wie viele innere Observables zeitgleich abonniert werden sollen. Ist dieser Wert nicht gesetzt, wird der Standardwert von unendlich genommen. Wird der Wert 1 angegeben, ist der Operator identisch zu concatMap. In diesem Fall abonniert flatMap jeweils nur ein inneres Observable zur selben Zeit und nimmt sich erst dann das nächste vor, wenn das vorherige abgeschlossen ist. Dies ist besonders praktisch, wenn zum Beispiel mehrere HTTP-Anfragen gemacht werden müssen, die voneinander abhängen.

switchMap

switchMap funktioniert ähnlich wie flatMap. Es wird ebenfalls eine Funktion erwartet, welche ein Observable zurückgibt. Im Gegensatz zu flatMap leitet switchMap nicht alle Ereignisse der inneren Observables an den Beobachter weiter, sondern lediglich diejenigen, die vom zuletzt ausgegebenen äußeren Observable ausgegeben worden sind. Das folgende Beispiel zeigt dies: Ein IntervalObservable gibt in Abständen von 400 ms seinen Index aus und wird mit switchMap auf je ein Observable gemappt, welches 10 mal einen modifizierten String zurückliefert. Es wird künstlich verzögert um innerIndex * 100 ms. Nach je 400 ms werden die Abonnements der inneren Observables storniert, da das äußere Observable ein neues Ereignis ausgibt. Es werden also nur die inneren Observables beim Beobachter ankommen, die innerhalb der 400-Millisekunden-Grenze ausgegeben werden (außer bei den inneren Observables, die durch das zuletzt ausgegebene Ereignis des äußeren Observables erzeigt wurden).

forkJoin

Der forkJoin Operator ist vergleichbar mit Promise.all. Er akzeptiert als Argument entweder mehrere Observables oder ein Array von Observables. Forkjoin abonniert dann all diese übergebenen, inneren Observables parallel und wartet auf deren Abschluss. ForkJoin gibt ein Observable zurück, welches die Ergebnisse der inneren Observables als Array in der Reihenfolge enthält, wie die inneren Observables an den forkJoin Operator übergeben worden sind. Dieses Array enthält jedoch nur das zuletzt ausgegebene Ereignis der inneren Observables. ForkJoin eignet sich also nicht für Observables, die mehr als einen Wert ausgeben. Zu beachten: Löst ein inneres Observable einen Error aus, ohne dass dieser mittels .catch abgefangen wird, so bricht forkJoin die gesamte Operation ab und leitet den Fehler an die OnError-Funktion des ForkJoin-Beobachters weiter.

combineLatest

combineLatest akzeptiert als Argumente ebenfalls ein Array von Observables oder mehrere Observables. Dieser Operator abonniert alle übergebenen inneren Observables, und wartet darauf, dass jedes einmal einen Wert ausgibt. Ab diesem Moment kombiniert combineLatest die jeweils zuletzt ausgegebenen Werte der inneren Observables in einem Array, welches an den Beobachter von combineLatest ausgegeben wird. Wie das Kugeldiagramm in Abb. 1 zeigt, gibt combineLatest immer dann das kombinierte Array aus, wenn ein inneres Observable einen neuen Wert ausgibt. Mit diesem Operator können beispielsweise sehr praktisch verschiedene DOM Events kombiniert werden wie die x/y Koordinaten der Mausposition in Verbindung mit dem aktuellem pageYOffset.

Abbildung 1: combineLatest-Diagramm

Grafikquelle: reactivex.io/rxjs/class/es6/Observable.js~Observable.html

timer und interval

Observable.timer und Observable.interval sind Observable Wrapper für die nativen Javascript Funktionen setTimeout und setInterval.

Timer ermöglicht das Ausführen einer Aktion nach einem bestimmten Zeitraum oder zu einem definierten Zeitpunkt. Der Timer Operator akzeptiert drei Parameter: die initiale Verzögerung in Millisekunden oder ein Ausführungszeitpunkt als Date-Objekt. Nach Ablauf der Verzögerung oder bei Erreichen des Zeitpunktes gibt das Timer Observable ein Ereignis aus. Argument 2 bestimmt das Ausführungsintervall für alle nachfolgenden Ereignisse. Ist dieser Wert leer, so werden keine weiteren Ereignisse ausgegeben. Für den dritten Operator, der nur selten notwendig sein dürfte, verweise ich auf die offizielle Dokumentation.

Der Interval Operator gibt jeweils ein Ereignis aus nach dem Ablauf der gegebenen Zeit in Millisekunden. Der Interval Operator kann durch den Timer Operator ersetzt werden: Observable.interval(1000) = Observable.timer(1000,1000).

Abschließende Einschätzung

Ich bin Angular dankbar dafür, dass ich gezwungen worden bin, mich mit RxJS auseinanderzusetzen. Die in diesem Blogpost vorgestellten Beispiele zeigen nur die Basics und essentiellen Möglichkeiten, die man mit Observables so anstellen kann. Seine Stärken entfaltet die Library erst dann, wenn die verschiedenen Operatoren kombiniert werden. Solche Code-Anweisungen können dann zwar sehr lang werden, sind aber, wie es für das funktionale Programmierparadigma typisch ist, noch deutlich kürzer als komplett selbst geschriebener Code.

Angular bietet für HTML-Templates die Async-Pipe an. Mit ihr ist es möglich, Promises und vor allem auch Observables auszuwerten und dann die Ergebnisse so zu verwenden, als wären es synchrone Aufrufe aus der Komponente. Grundsätzlich wäre es also möglich, nahezu alle Datenstrukturen, die irgendwo verwendet werden, in Observables zu verpacken. Das ermöglicht nicht nur durchgehend die Nutzung der wirklich praktischen Observable-Operatoren, sondern hätte auch den Vorteil, dass sämtliche Funktionen asynchron ausgeführt werden würden.

Der Einstieg in RxJS ist zwar erst einmal schwer. Nicht zuletzt aufgrund der vielen Operatoren habe ich anfangs lieber weiterhin das altbewährte Promise verwendet. Inzwischen habe ich fast alle Funktionen, die ein Promise zurücklieferten auf Observables umgestellt. Wer einmal das Prinzip und die Vorteile erkannt hat, wird RxJS wohl nie wieder missen wollen.