Jak mogę sprawdzić, co robi moje zapytanie o reaktywne rozszerzenia?

Question

Piszę złożone zapytanie o Reactive Extensions z dużą liczbą operatorów. Jak mogę zobaczyć, co się dzieje?

Pytam i odpowiadanie na to pytanie, ponieważ pojawia się ono dość często i prawdopodobnie ma dobry ogólny użytek.

Answer 1

Możesz dodać tę funkcję liberalnie do operatorów Rx podczas ich rozwój, aby zobaczyć, co się dzieje:

public static IObservable<T> Spy<T>(this IObservable<T> source, string opName = null) 
    { 
     opName = opName ?? "IObservable"; 
     Console.WriteLine("{0}: Observable obtained on Thread: {1}", 
          opName, 
          Thread.CurrentThread.ManagedThreadId); 

     return Observable.Create<T>(obs => 
     { 
      Console.WriteLine("{0}: Subscribed to on Thread: {1}", 
           opName, 
           Thread.CurrentThread.ManagedThreadId); 

      try 
      { 
       var subscription = source 
        .Do(x => Console.WriteLine("{0}: OnNext({1}) on Thread: {2}", 
               opName, 
               x, 
               Thread.CurrentThread.ManagedThreadId), 
         ex => Console.WriteLine("{0}: OnError({1}) on Thread: {2}", 
               opName, 
               ex, 
               Thread.CurrentThread.ManagedThreadId), 
         () => Console.WriteLine("{0}: OnCompleted() on Thread: {1}", 
               opName, 
               Thread.CurrentThread.ManagedThreadId) 
        ) 
        .Subscribe(obs); 
       return new CompositeDisposable(
        subscription, 
        Disposable.Create(() => Console.WriteLine(
          "{0}: Cleaned up on Thread: {1}", 
          opName, 
          Thread.CurrentThread.ManagedThreadId))); 
      } 
      finally 
      { 
       Console.WriteLine("{0}: Subscription completed.", opName); 
      } 
     }); 
    } 

Oto Wykorzystanie przykład pokazuje, subtelna różnica zachowanie Range:

Observable.Range(0, 1).Spy("Range").Subscribe(); 

Daje moc wyjściową:

Range: Observable obtained on Thread: 7 
Range: Subscribed to on Thread: 7 
Range: Subscription completed. 
Range: OnNext(0) on Thread: 7 
Range: OnCompleted() on Thread: 7 
Range: Cleaned up on Thread: 7 

Ale to:

Observable.Range(0, 1, Scheduler.Immediate).Spy("Range").Subscribe(); 

Daje wyjście:

Range: Observable obtained on Thread: 7 
Range: Subscribed to on Thread: 7 
Range: OnNext(0) on Thread: 7 
Range: OnCompleted() on Thread: 7 
Range: Subscription completed. 
Range: Cleaned up on Thread: 7 

Spot różnica?

Oczywiście można zmieniać tego napisać do dzienników lub do debugowania, lub użyć dyrektywy preprocesora zrobić chude subskrypcji pass-through na kompilacji Release etc ...

Można zastosować Spy całym łańcuchu operatorów . np .:

Observable.Range(0,3).Spy("Range") 
      .Scan((acc, i) => acc + i).Spy("Scan").Subscribe(); 

Daje wyjście:

Range: Observable obtained on Thread: 7 
Scan: Observable obtained on Thread: 7 
Scan: Subscribed to on Thread: 7 
Range: Subscribed to on Thread: 7 
Range: Subscription completed. 
Scan: Subscription completed. 
Range: OnNext(1) on Thread: 7 
Scan: OnNext(1) on Thread: 7 
Range: OnNext(2) on Thread: 7 
Scan: OnNext(3) on Thread: 7 
Range: OnCompleted() on Thread: 7 
Scan: OnCompleted() on Thread: 7 
Range: Cleaned up on Thread: 7 
Scan: Cleaned up on Thread: 7 

Jestem pewien, że można znaleźć sposoby na wzbogacenie to do własnych celów.