Opisywałem już mechanizm konwersji zdarzeń .NET do IObservable. W RX istnieje dodatkowo nowy mechanizm, mający na celu zastąpić standardowe zdarzenia  .NET – przynajmniej w części przypadków. Zastanówmy się, co jest złego tak naprawdę w obsłudze zdarzeń w .NET?

1. Dość często programiści zapominają usunąć zdarzenie co może skutkować memory leak. Czasami jest ciężko zdefiniować moment, w którym należy usunąć zdarzenie. Pewnym rozwiązaniem problemu może być zastosowanie wzorca weak event pattern ale jak wiemy z archiwalnych wpisów, nie jest to też takie proste.

2. Zdarzenia, które zawierają anonimowe metody są szczególnie trudne do usunięcia. Na przykład:

internal class Program
{
   public static event EventHandler MessageReceived;
   private static void Main(string[] args)
   {
       MessageReceived += (sender,e) => Console.WriteLine(e.ToString());
   }
}

W jaki sposób usunąć powyższe zdarzenie? Niestety trzeba stworzyć tymczasową zmienną i ją przechowywać gdzieś:

public static event EventHandler MessageReceived;

private static void Main(string[] args)
{
  EventHandler action=(sender, e) => Console.WriteLine(e.ToString());

  MessageReceived += action;
  // disposing
  MessageReceived -= action;
}

3.  Jak już wspomniałem i pokazałem w jednym z poprzednich wpisów, nie ma możliwości odczytania poprzednich wartości, jakie zostały wygenerowane przez zdarzenie (np. poprzednia pozycja myszki).

4. Zdarzenia to właściwości a nie obiekty co czasami może być niewygodne.

5. Z punktu pisania testów jednostkowych, zdarzenia również nie są szczególnie łatwe do obsługi.

6. Niestety zdarzeń nie można traktować jako kolekcji, z tego względu nie ma możliwości wykorzystania zapytań  LINQ.

Dość narzekania, czas pokazać mechanizm jaki dostarcza RX. Mam nadzieję, że wszyscy są już obeznani z wzorcem projektowym Observer. Jeśli nie, polecam poczytanie o nim( np. na tym blogu, kiedyś o tym już pisałem). Znając zasady działania wzorca obserwator, mechanizm działania zdarzeń RX nie powinien zaskakiwać. Najważniejszym interfejsem jest ISubject, który wygląda następująco:

public interface ISubject<in TSource, out TResult> 
: IObserver<TSource>, IObservable<TResult>
{ 
}
public interface ISubject<T>  : ISubject<T, T>
{ 
}

Proszę zauważyć, że implementuje on zarówno IObserver jak IObservable. W praktyce oznacza to, że klasy implementujące ISubject, mogą zarówno publikować jak i dokonywać subskrypcji (nasłuchiwać). Przykład:

internal class Program
{
   public static  Subject<string> MessageReceived=new Subject<string>();

   private static void Main(string[] args)
   {            
       MessageReceived.Subscribe(Console.WriteLine);
       MessageReceived.OnNext("Hello World");
       Console.ReadLine();
   }
}

W zwykłych zdarzeniach, operator += służy do podłączenia metod. W RX analogiczną funkcję pełni metoda Subscribe. Zasada działania jest taka sama, jak w przypadku dobrze znanego IObservable ( w końcu ISubject implementuje IObservable). W celu wywołania zdarzenia, korzystamy analogicznie z OnNext. Dzięki RX, łatwo zwolnić zdarzenie:

internal class Program
{
   public static  Subject<string> MessageReceived=new Subject<string>();

   private static void Main(string[] args)
   {            
       IDisposable disposable=MessageReceived.Subscribe(Console.WriteLine);
       MessageReceived.OnNext("Hello World");
       disposable.Dispose();
       MessageReceived.OnNext("You will not see this text");

       Console.ReadLine();
   }
}

Oczywiście, można (a nawet jest to polecane) użyć klauzuli using, dla powyższego przykładu.

internal class Program
{
   public static  Subject<string> MessageReceived=new Subject<string>();

   private static void Main(string[] args)
   {            
       using(MessageReceived.Subscribe(Console.WriteLine))
       {
           MessageReceived.OnNext("Hello World");
       }

       MessageReceived.OnNext("You will not see this text");

       Console.ReadLine();
}

Zdarzenie automatycznie zostanie zwolnione (a wszelkie metody połączonego do niej, usunięte) kiedy zmienna nie ma żadnych referencji. Przypomina to trochę weak event pattern, którego cel był analogiczny – ale wtedy należało wykonać dużo więcej roboty. Dzięki RX mamy możliwość większej kontroli nad przepływem zdarzeń za pomocą metod OnNext, OnError, OnCompleted:

internal class Program
{
   public static Subject<string> MessageReceived = new Subject<string>();

   private static void Main(string[] args)
   {
       using (MessageReceived.Subscribe(OnNext,OnCompleted,OnError))
       {
           MessageReceived.OnNext("Processing...");
           try
           {
               // do sth
           }
           catch (Exception e)
           {
               MessageReceived.OnError(e);
           }
           finally
           {
               MessageReceived.OnCompleted();
           }
       }

       Console.ReadLine();
   }
   private static void OnNext(string obj)
   {
   }
   private static void OnCompleted(Exception error)
   {
   }
   private static void OnError()
   {
   }
}

W przypadku standardowych zdarzeń, nie ma oczywiście takiej możliwości. Jeśli ktoś jest zaznajomiony z RX to Subject wygląda bardzo prosto – implementuje on interfejsy, które były bardzo dokładnie omówione w poprzednich wpisach.