Die Power von Rx für Event-Based Components

Ob Event-Based Components (EBC) auch mit den Reactive Extensions, dem Rx Framework von Microsoft implementiert werden könnten, stand als Frage schon länger im Raum. Jetzt hab ich mich mal daran versucht.

Als Beispiel habe ich die FizzBuzz Kata gewählt. Der EBC-Entwurf ist denkbar simpel:

Bisher wurde dann eine EBC-Aktion wie Zahl transformieren so übersetzt:

   1: class Zahl_transformieren_früher

   2: {

   3:     public void In_Zahl(int zahl)

   4:     {

   5:         ...

   6:     }

   7:  

   8:     public event Action<string> Out_Symbol;

   9: }

Output-Pins von EBC-Aktionen waren Events, Input-Pins Methoden. Das ist eine geradlinige Übersetzung. Funktioniert in der Praxis tadellos.

Und doch bleibt etwas zu wünschen übrig. Es gibt einfach keine Unterstützung für Operationen auf Folgen von Events, also Nachrichtenströme. Natürlich kann man Aktionen wie Filter in einen Draht “einklemmen” – doch die müssen relativ umständlich erst entwickelt werden.

Mit Rx geht das jedoch viel einfacher. Damit kann man einen Event-Strom filtern oder verzögern oder zwei Ströme zusammenführen usw. usf. Viele schöne Operationen auf Events – oder eben Observables – gibt es dort. Warum das nicht für EBC nutzen?

Deshalb hier mein Vorschlag für eine Übersetzung von EBC-Flows mit Rx-Mitteln:

* Ein Output-Pin wird in ein IObservable<T> übersetzt
* Ein Input-Pin wird in einen IObserver<T> übersetzt

Mit diesen Interfaces kann man dann wunderschöne Sachen machen.

Aber es sind eben nur Interfaces und so brauchen wir noch Implementationen. Ich nenne die mal OutputPin<T> und InputPin<T>. Ist doch irgendwie naheliegend, oder?

   1: class Zahl_transformieren

   2: {

   3:     public InputPin<int> In_Zahl { get; private set; }

   4:     public OutputPin<string> Out_Symbol { get; private set; }

   5:  

   6:     public Zahl_transformieren()

   7:     {

   8:         this.In_Zahl = new InputPin<int>(Transform);

   9:         this.Out_Symbol = new OutputPin<string>();

  10:     }

  11:  

  12:  

  13:     private void Transform(int zahl)

  14:     {

  15:         if (IsFizzBuzz(zahl))

  16:             this.Out_Symbol.Post("fizzbuzz");

  17:         ...

  18:     }

Um die eigentlichen Transformationsroutine Transform() wird nun ein IObserver<T> gewickelt in Form eines InputPin<T>. Und die Ergebnisse werden an einen OutputPin<T> geschickt, der ein IObservable<T> ist. Den können dann folgende Aktionen abonnieren.

Die Zahlengenerierung sieht analog aus. Kein Hexenwerk. Beide Aktionen verdrahte ich dann noch in einer Aktionsgruppe zu einem Flow, um die Details der Verarbeitung gegenüber einem Client zu verbergen:

   1: public class FizzBuzzer

   2: {

   3:     public InputPin<Unit> In_Generate { get; private set; }

   4:     public OutputPin<string> Out_Symbole { get; private set; }

   5:  

   6:  

   7:     public FizzBuzzer()

   8:     {

   9:         var ze = new Zahlen_erzeugen();

  10:         var zt = new Zahl_transformieren();

  11:  

  12:         this.In_Generate = ze.In_Generate;

  13:         ze.Out_Zahl.Subscribe(zt.In_Zahl);

  14:         this.Out_Symbole = zt.Out_Symbol;

  15:     }

  16: }

Dem Input-Pin der Aktionsgruppe kann der Input-Pin von Zahlen erzeugen zugewiesen werden. Sie braucht also keinen eigenen. Dito für den Output-Pin der Aktionsgruppe.

Und die konsumierende Aktion abonniert die produzierende Aktion, s. Zeile 13. Das entspricht der bisherigen Zuweisung einer Input-Pin Methode als Eventhandler an einen Output-Pin Event.

Formal ist der Unterschied zwischen bisheriger Übersetzung und der nach Rx minimal, finde ich. Das setzt sich dann bis zum Client fort, der die Aktionsgruppe nutzt:

   1: var fb = new FizzBuzzer();

   2:  

   3: fb.Out_Symbole.Subscribe(Console.WriteLine);

   4: fb.In_Generate.OnNext(new Unit());

Ein kleinwenig gewöhnungsbedürftig mag allerdings sein, dass Rx keine Kommunikation ohne Wert kennt. Deshalb muss die Zahlengenerierung mit einem Parameter angestoßen werden. Dankenswerterweise gibt es dafür Unit, sozusagen einen “wertlosen” Typ. Er ist der Funktionalen Programmierung entlehnt und findet sich auch in F# wieder.

Und jetzt zum Gewinn durch eine Umstellung der Übersetzung von EBC mit Rx:

   1: fb.Out_Symbole

   2:   .Where(s => s[0] < '@')

   3:   .Select(s => int.Parse(s))

   4:   .SkipWhile(i => i < 42)

   5:   .Subscribe(Console.WriteLine);

Wir können ganz leicht in den “Eventfluss” eingreifen. Es gibt viele Standardoperationen dafür, allen voran die bekannten und beliebten Linq-Operatoren.

Wie klingt das? Ich finde das sehr vielversprechend. Damit experimentiere ich weiter. Mein Gefühl ist, dass mit Rx die Übersetzung noch systematischer wird und gleichzeitig die Möglichkeiten zum Umgang mit Event-Strömen wachsen. Also ein doppelter Gewinn.