Auf Listen oder einfachen Array ist man es inzwischen schon gewohnt: Anstelle von einfachen Schleifen, bedient man sich hier lieber einem Iterator, oder verfällt ganz und gar dem Funktionalen Ansatz. Aber warum eigentlich? Gerade der funktionale Ansatz liefert eine einheitliches Interface wie man mit Collections umgeht. Den selben Effekt erreiche ich aber auch mit Iteratoren. Eine Liste von Cocktails

var cocktails = [
    {id: 1001, name: 'Piña Colada', zutaten: [], prozent: 5.0 },
    {id: 1002, name: ' Tequila Sunrise', zutaten: [], prozent: 6.0 },
    {id: 1003, name: ' Long Island', zutaten: [], prozent: 7.0 },
];

lässt sich beispielsweise so nach dem id/name Paar für Cocktails mit mehr als 6% durchsuchen:

var newList = [];
for(var i = 0; i <= cocktails.length; i++) {
    if (cocktails[i].prozent > 6.0) {
        newList.push({id: cocktails[i].id, title: cocktails[i].title})
    }
}

Das erfüllt völlig seinen Zweck, aber schön ist das nicht. Mit einer foreach-Funktion könnte das dann so ausshen:

var newList = [];
cocktails.forEach(function (cocktail) {
    if (cocktails[i].prozent > 6.0) {
        newList.push({id: cocktail.id, title: cocktail.title})
    }
});

aber noch besser wäre:

var godOnes = cocktails
    .filter(function (cocktail) {
        return cocktail.prozent > 6.0;
    })
    .map(function (cocktail) {
        return {id: cocktail.id, name: cocktail.name};
    });

Stream von Events

Wie wäre es, wenn man ein Stream von Events auf die gleich Art und Weise durchlaufen könnte? Wenn es egal ist, ob ich eine Liste von Einträge mit map() zurecht stutze, oder mir nur die Key-Up-Events von Tastenanschlägen in einem Input-Field mit einem filter() heraus suche, die mich interessieren? Wenn ich mich dann noch auf diesem Stream von Events mit einem Observer registrieren könnte, könnte ich durch Events viele Teile meiner Applikation entkoppeln.

Reactive Extensions

Genau das geben uns die Reactive Extensions. In den 2000ern von Erik Meier und Mathew Podwysocki bei Microsoft entwickelt, waren sie eigentlich auf der Suche nach einer Lösung, um Applikationen plattformunabhängig kompilieren zu können. Einer der Übergang von einer Plattform zu einer anderen bedeutete, Formulare in der Windows-Umgebung in Formulare einer Web-Applikation zu übersetzen. Doch das Web ist in höchsten Maße asynchron. Was bedeutete das? Das Microsoft-Team musste einen Weg finden diese Entkoppelung zu abstrahieren. Heraus kam das in Microsoft-Kreisen auch „LINQ to Events“ genannte Konstrukt. Denn LINQ (Language Integrated Query) war schon ein Weg in .NET, jede Ansammlung von Daten mit dem selben Interface wie Collections zu behandeln. Man kann sich das vorstellen wie ein SQL Query auf jede Form von Liste, Collection, Set oder Array.

Die Reactive Extensions definieren dabei nur ein Interface, das inzwischen schon in vielen Sprachen implementiert wurde. Angefangen natürlich bei .NET, über Javascript (RxJs) bis hin zu Java (RxJava) haben alle Plattformen ihre eigene Implementierung. Zusammengefasst kann man sie in der Github-Organisation finden. Selbst eine PHP-Implementierung gibt es, leider scheint diese nicht in der Hauptorganisation gelistet zu sein

Reactive Javascript

Reactive Javascript (RxJS) ist nun ein Teil dieser Reactive Extensions. Da größtenteils vom Mathew Podwysocki entwickelt, ist es sehr nahe an der .NET-Implementierung, was es um so interessanter macht. Schauen wir uns einmal ein einfaches Beispiel an:

var source = Rx.Observable.range(1,10);

var reducedSource = source.filter(function (value) {
    return value % 2 === 0;
});

var subscription = reducedSource.subscribe(
  function (x) { console.log('onNext: %s', x); },
  function (e) { console.log('onError: %s', e); },
  function () { console.log('onCompleted'); }
  );

subscription.dispose();

Da sich Events und deren Streams in Textform erst einmal schlecht darstellen lassen, soll hier einfach mal eine Liste von Zahlen (1-10) her halten. Mit Rx.Observable.range(x, y) kann ich ein Observable erstellen, das genau eine solche Liste enthält. Auf diesem kann sich jetzt entweder ein Observer subscriben, oder man wendet Operatoren an, um den Stream entweder zu verändern (Projektionen), ergänzen (Kombinatoren), oder zu reduzieren (Filter). Der Observer entspricht dann den 3 Callback-Funktionen, die ich reducedSource.subsribe() übergebe. Interessant ist hier, dass man sich von so einem Stream auch unsubscriben kann. Die subscribe() Methode gibt hier ein sogenanntes Disposal zurück, auf dem ich mich von einem Stream auch wieder lösen kann: dispose(). Die Operatoren erzeugen wieder neue Observables, die sich nicht gegenseitig beeinflussen:

var subscription = reducedSource.subscribe(
  function (x) { console.log('Reduced Source: %s', x); }
  );

var subscription = source.subscribe(
  function (x) { console.log('Source: %s', x); }
  );

liefert mir:

> ReducedSource: 2
> ReducedSource: 4
> ReducedSource: 6
> ReducedSource: 8
> ReducedSource: 10
> Source: 1
> Source: 2
...
> Source: 10

Das heißt, man kann sich hier die Streams auf den jeweiligen Anwendungsfall zurecht biegen.

Damit es auch ein wenig interessanter wird, kann man natürlich nicht nur Listen von Zahlen bearbeiten, sondern auch Events und Callbacks:

Rx.Observable.fromEvent(element, eventName, [selector]);
Rx.Observable.fromCallback(func, [context], [selector]);

Damit könnte man dann beispielsweise einfach den Tastenanschlag in einem Input-Field abfangen:

var input = $('#input');

var source = Rx.Observable.fromEvent(input, 'keyup');

var subscription = source.subscribe(
  function (x) {console.log('Next: key pressed!');},
  function (err) {console.log('Error: %s', err);},
  function () {console.log('Completed');});

Neben dem oben gezeigten filter() Operator gibt es noch viele weitere. Am besten kann man diese in den RxMarbles sehen, da sie hier gleich visualisiert werden.

Array vs. Observables

Wenn wir uns die Operatoren anschauen, kann man sich die berechtigte Frage stellen: Worin liegt nun der Unterschied zwischen Array-Operatoren, wie man sie inzwischen in vielen Libraries findet (z.B. Underscore) und Operatoren, die man auf Observables anwenden kann?

Operatoren auf Arrays nehmen immer ein ganzes Array auf und geben auch hinten wieder ein verändertes (reduziert, gefiltert, ..) Array wieder aus. Das heisst es wird immer wieder auf ganzen Arrays gearbeitet, die Item für Item durch gegangen werden. Bei Streams von Events ist das anders. Hier kommt Event für Event in die Kette von Operatoren rein. Ein Filter, der ein Event ablehnt, gibt auch nichts zurück. Das heisst hier wird der Durchlauf durch die Kette sofort unterbrochen. Dazu kann ich den Stream jederzeit abbestellen, was mir bei Array-Operatoren schwer fallen würde.

Conclusion

Wen das Thema tiefer interessiert, der kann sich hier viele hilfreiche Tutorials anschauen. Dazu gibt es auch ein Online-Buch. Es beschreibt die Reactive Extensions zwar in C#, aber das macht hier kaum einen Unterschied. Die Javascript Dokumentation findet man im Github. Und nun viel Spaß beim Probieren. Ein kleines Beispiel zu meinem Vortrag zu dem Thema findet man im jsFidle.