[RX 2] Kolekcje, to podstawa, czyli wprowadzenia do Reactive Extensions część 2 [PL]

W ramach kontynuacji tematyki związanej z Reactive Extensions (RX)przyjrzyjmy się elementom, które stoją u podstaw ich działania, czyli kolekcjom i wzorcowi Obserwator.

W poprzedniej części artykułu poznaliśmy cel przyświecającym twórcom Reactive Extensions (RX) jak i skąd można pobrać wspomniane rozszerzenia. Skoro już mamy zainstalowane Reactive Extensions (Rx), zacznijmy przyglądać się jego możliwością.

Jak wcześniej zostało wspomniane, Rx upraszcza programowanie asynchroniczne, pozwala na łatwe filtrowanie czy komponowanie zdarzeń (np. z wykorzystaniem LINQ), a wszystko to dzięki obserwowanym kolekcją. Przyjrzyjmy się więc temu najbardziej elementarnemu elementowi, czyli kolekcjom. Oczywiście, to czym są kolekcje chyba każdy wie, jednak nie każdy skojarzy od razu z nimi interfejsy IEnumerable i Ienumerator. Stanowią one implementację wzorca projektowego Iterator w celu przeglądania owych kolekcji.

Deklaracje tych interfejsów wyglądają następująco:

  public interface IEnumerable<T>
  {
    IEnumerator<T> GetEnumerator();
  }

  public interface IEnumerator<T>: IDisposable
  {
    T Current { get; }
    bool MoveNext();
    void Reset();
  }

Interfejs IEnumerable służy do pobrania instancji iteratora wspierającej interfejs IEnumerator, natomiast IEnumerator pozwala na iterację poszczególnych elementów listy, w szczególności na pobranie aktualnego elementu (właściwość Current), przejście do następnego elementu (funkcja MoveNext) i powrót do początku listy (Reset). Jest to klasyczny przykład synchronicznego dostępu do danych, najpierw prosimy kolekcję o iterator (IEnumerable), a później błagamy o każdą kolejną daną (MoveNext). Warto tu podkreślić, że na czas pobierania lub przesuwania do kolejnego elementu jesteśmy zablokowani w oczekiwaniu na wykonanie operacji. Interfejsy te są stosowane w .NET dość pospolicie, zaczynając na słowie kluczowym foreach (które pobiera właśnie wspomniany iterator, by umożliwić przeglądanie całej listy) lub idąc w kierunku bardziej zaawansowanej maszyny związanej z LINQ (która właśnie bazuje na wspomnianych interfejsach i możemy dzięki temu stosować operatory takie jak: Where Orgderby, GroupBy, Select). Warto tutaj zwrócić uwagę, że jest to przykład wyciągania (PULL) danych z kolekcji (ze źródła danych). Nie muszę chyba tutaj nikogo przekonywać, że nie jest to optymalne podejście, dużo lepszym rozwiązaniem byłoby odbieranie wysyłanych (wpychanych – PUSH) przez kolekcję danych.

W przyrodzie na każdą siłę działa kontr-siła, przeciwieństwa się niwelują lub inaczej mówiąc uzupełniają. Dane mogą być interaktywnie wyciągane ze źródła danych lub źródło danych może nam je wpychać, a my musimy tylko odpowiednio reagować na otrzymywane dane. Jest to przejście z programowania interaktywnego na reaktywne. Można by powiedzieć, że są to podejścia dualne.
Z matematycznego punktu widzenia, jeżeli mamy pewne prawo, można przy pomocy pewnych przekształceń przejść do prawa mu dualnego (przypomnijmy sobie np. prawa De Morgana i przekształcenia oparte o algebrę Boole'a). Spróbujmy więc przekształcić wspomniane interfejsy IEnumerable i IEnumerator do interfejsów im dualnych (i w pewien sposób przeciwnych).

Przekształcenie to można znaleźć m.in. na prezentacji Bart de Smet na Channel 9 p.t. (Rx: Curing your asynchronous programming blues).

Uwaga: dla ułatwienia miejsca ostatnich zmian i przekształceń zostają oznaczone kolorem czerwonym.

Zacznijmy więc od zauważenia, że jeżeli funkcja nie przyjmuje żadnych parametrów, to równie dobrze można tutaj zaznaczyć, że przyjmuje ona parametr typu pustego (void). W tym kroku zmieniona zostaje właściwość Current na funkcję GetCurrent, która udostępnia tą samą funkcjonalność. Wykreślona zostaje funkcja Reset jako nieistotna z punktu widzenia tego przekształcenia. Otrzymujemy więc:

  public interface IEnumerable<T>
  {
    IEnumerator<T> GetEnumerator(void);
  }

  public interface IEnumerator<T>: IDisposable
  {
    T GetCurrent (void);
    bool MoveNext(void);
    void Reset();
  }

W następnym kroku zauważmy, że skoro IEnumerator jest IDisposable, to tak naprawdę funkcja GetEnumerator zwraca instancję spełniającą również interfejs IDisposable, przenieśmy więc tą informację do funkcji GetEnumerator. Zauważmy również że funkcja MoveNext może zwrócić wyjątek, co (przy pomocy konwencji podobne do znanej z języka Java) zostało zaznaczone jako „throws Exception”).

  public interface IEnumerable<T>
  {
    (IDisposable & IEnumerator<T> GetEnumerator(void);
  }

  public interface IEnumerator<T>: IDisposable
  {
    T GetCurrent (void);
    bool MoveNext(void) throws Exception;
 }

Przejdźmy teraz do interfejsu dualnego dla interfejsu IEnumerable, będzie on roboczo nazwany IEnumerableDual, funkcja GetEnumerator, została zmieniona w nim w SetEnumerator i przyjmuje jako argument IEnumerator). Dla interfejsu IEnumerator i funkcji MoveNext zrezygnujmy więc ze znanego z języka Java dopisku „throws Excetion” i zapiszmy to tak, jak się powinno, czyli napiszmy, że funkcja zwraca Exception.

  public interface IEnumerableDual<T>
  {
    (IDisposable & IEnumerator<T>) GetSetEnumerator( IEnumerator<T>);
  }

  public interface IEnumerator<T>
  {
    T GetCurrent (void);
    (bool | Exception) MoveNext(void) throws Exception;
 }

W następnym kroku zauważamy, że typ bool to po prostu wartość logiczna true lub false, mamy więc:

  public interface IEnumerableDual<T>
  {
    IDisposable SetEnumerator( IEnumerator<T>);
  }

  public interface IEnumerator<T>
  {
    T GetCurrent (void);
    (true | false | Exception) MoveNext(void);
 }

Poświęćmy jeszcze trochę uwagi interfejsowi IEnumerator, chyba jest coś w nim nadmiarowego, poza tym funkcja, dziwna jest funkcja, która zawsze zwraca wartość logiczną true. Może lepiej zrezygnować z funkcji GetCurrent i niech funkcja MoveNex zwraca zamiast informacji o powodzeniu nową wartość.

  public interface IEnumerableDual<T>
  {
    IDisposable SetEnumerator( IEnumerator<T>);
  }

  public interface IEnumerator<T>
  {
    T GetCurrent (void);
    (T| false | Exception) MoveNext(void);
 }

Przejdźmy teraz do interfejsu dualnego dla IEnumerator, czyli IEnumeratorDual. Parametry wyjściowe funkcji MoveNext przesuwamy na „wejście”, natomiast nazwę funkcji zmieniamy na SetNext:

  public interface IEnumerableDual<T>
  {
    IDisposable SetEnumerator( IEnumerator<T>);
  }

  public interface IEnumeratorDual<T>
  {
     void MoveSetNext(T| false | Exception);
  }

Oczywiści, gdy jakaś funkcja przyjmuje różne argumenty na wejściu, to tak naprawdę mamy kilka definicji tej funkcji. Pokazane zostało to poniżej, z jednoczesną zamianą nazw poszczególnych funkcji na bardziej trafne:

  public interface IEnumerableDual<T>
  {
    IDisposable SetEnumerator( IEnumerator<T>);
  }

  public interface IEnumeratorDual<T>
  {
    void SetValue(T);
    void SetException (Exception);
    void SetNothing ();
 }

Powyższe przekształcenie możemy uznać za skończone, otrzymaliśmy interfejsy dualne do IEnumerable i IEnumerator. Gdy się im przyjrzymy dokładniej, okaże się że ich definicja jest zgodna z definicjami klas/interfejsów bazowych niezbędnych do realizacji wzorca obserwator:

W naszym przekształceniu potrzebujemy jeszcze tylko zmienić nazwy interfejsów i nazwy ich funkcji, by ostatecznie otrzymać, zgodne z powyższym rysunkiem definicje:

 public interface IObservable<T>
  {
    IDisposable Subscribe( IObserver<T>);
  }

  public interface IObserver<T>
  {
    void OnNext(T);
    void OnException (Exception);
    void OnCompleted ();
 }

Poznajmy IObservable i IObserver dwa interfejsy pozwalające na realizację wzorca projektowego obserwator. IObserver (zwany w niektórych książkach poświęconych wzorcom projektowym Subject'em) pozwala na podłączanie się (Subscribe lub czasami w innych implementacjach Attach) i odłączanie się (poprzez wykorzystanie interfejsu IDisposable lub czasami w innych implementacjach Dettach). Podczas podłączenia przekazujemy instancję wspierającą interfejs Iobserwer, czyli obserwator, który ma za zadanie obserwować zmiany. Wspomniany obserwator (IObserver) jest powiadamiany (Notiffy) o:

• następnej/nowej wartości (OnNext),
• błędzie, jeżeli jakiś wystąpił (OnError),
• końcu, jeżeli obserwacja powinna się zakończyć (OnComplete).

Zauważmy również, że dzięki temu dane są do nas „wpychane”, nie musimy o każdą daną prosić, otrzymujemy „asynchroniczny PUSH”. 

Udowodniliśmy więc, że interfejsy Iterator i Obserwator są wzajemnie dualne, co więcej realizacja ich pozwala przejść ze scenariusza wyciągania danych na otrzymywanie danych, co pozwala na przejście z komunikacji synchronicznej na asynchroniczną.

Mamy obserwowanie i subskrypcję zamiast przeglądania.

Wykorzystanie tego jest bardzo proste:

• Konstruujemy / pobieramy IObservable. (W Rx jest wiele metod pomocniczych tworzące IObservable z tablic, list, zdarzeń, …)
• Subskrybujemy, przekazując nasz obserwator (IObserver) lub odpowiednie delegacje.
• Wykonujemy „Dispose” na subskrypcji, której dłużej nie potrzebujemy.

Cdn. ... W następnej części artykułu będzie więcej przykładów związanych z samym Reactive Extensions (RX).