Reactive Framework Hello World

Question

This is an easy program to introduce Reactive Framework. But I want to try the error handler, by modifying the program to be:

var cookiePieces = Observable.Range(1, 10);
cookiePieces.Subscribe(x =>
   {
      Console.WriteLine("{0}! {0} pieces of cookie!", x);
      throw new Exception();  // newly added by myself
   },
      ex => Console.WriteLine("the exception message..."),
      () => Console.WriteLine("Ah! Ah! Ah! Ah!"));
Console.ReadLine();

In this sample this overload is used:

public static IDisposable Subscribe<TSource>(
     this IObservable<TSource> source, 
     Action<TSource> onNext, 
     Action<Exception> onError, 
     Action onCompleted);

I hoped I would see the exception message printed, but the console app crashed directly. I'm new to reactive framework, who can tell me the reason?

Answer 1

The exception handler is used for exceptions created in the observable itself, not by the observer.

An easy way to provoke the exception handler is something like this:

using System;
using System.Linq;

class Test
{
    static void Main(string[] args)
    {
        var xs = Observable.Range(1, 10)
                           .Select(x => 10 / (5 - x));

        xs.Subscribe(x => Console.WriteLine("Received {0}", x),
                     ex => Console.WriteLine("Bang! {0}", ex),
                     () => Console.WriteLine("Done"));

        Console.WriteLine("App ending normally");
    }
}

Output:

Received 2
Received 3
Received 5
Received 10
Bang! System.DivideByZeroException: Attempted to divide by zero.
   at Test.<Main>b__0(Int32 x)
   at System.Linq.Observable.<>c__DisplayClass35a`2.<>c__DisplayClass35c.<Select
>b__359(TSource x)
App ending normally

Answer 2

Does it really crash or jumps Visual Studio into and shows you that an exception happened? If the second is true, you should take a look into Debug - Exception within the menu bar and deselect everything on the right.

Answer 3

In the Rx library, any user code passed into an operator that works on IObservable (Select, Where, GroupBy etc...) will be caught and send to the OnError handler of observers subscribed to the observable. The reason these are handled is that they are part of the computation.

Exceptions occurring in Observer code will have to be handled by the user. As they're at the end of the computation, it is unclear to Rx how to handle these.