How can I find out what my request for active requests does?

Question

I am writing a complex Reactive Extensions request with many operators. How can I find out what is going on?

I ask and answer this because it fits in an honest cue ball and is probably well used.

Answer 1

You can add this function to your Rx statements while you design them to see what happens:

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); } }); } 

Here's a usage example showing the subtle difference in Range behavior:

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

Gives output:

 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 

But this:

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

Gives output:

 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 

Determine the difference?

Obviously, you can change this to write to logs or to Debug, or use preprocessor directives to do a lean end-to-end subscription to the build release, etc ...

You can apply Spy whole chain of statements. eg:.

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

Gives output:

 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 

I am sure that you can find ways to enrich this in accordance with your goals.