Reactive Extensions crash .net process

Question

I'm starting using reactive extensions in my project and found the following problem. I have a subject inside my class and I need that all client observers process OnNext messages in different threads.

private Subject<IEnumerable<int>> _multiResponse =
    new Subject<IEnumerable<int>>();

public IObservable<IEnumerable<int>>> MultiResponse
{
     get
     {
         return _response.ObserveOn(TaskPoolScheduler.Default);
     }
}

If the client throws exception in OnNext event all the process crashes because of unobserved exception in Task. StackTrace:

ProcessResponses(IEnumerable responses) in Scheduler.cs:line 149 at System.Reactive.AnonymousSafeObserver'1.OnNext(T value) at System.Reactive.ScheduledObserver'1.Dispatch(ICancelable cancel) at System.Reactive.Concurrency.Scheduler.b_32(Action'1 a, ICancelable c) at System.Reactive.Concurrency.TaskPoolScheduler.<>c_DisplayClass7'1.b__6() at System.Threading.Tasks.Task.Execute()

Is it an expected behavior or a bug in RX and what to do with it?

Answer 1

Right, paying attention now... this is about unobserved exceptions.

I think the fact that you are doing things asynchronously shouldn't matter. This synchronous code throws an Exception, as I would expect:

Observable.Return(1).Subscribe(
    _ => { throw new Exception(); });

However, I was hoping to find the following code would throw an Exception (thus explaining what you have seen) but at first I was very surprised to find that it does not!

Observable.Return(1).ObserveOn(TaskPoolScheduler.Default).Subscribe(
    _ => { throw new Exception(); });

(EDIT - see below - this is .NET 4.5 behaviour)

Certainly switching to the ThreadPoolScheduler gives the expected Exception:

Observable.Return(1).ObserveOn(ThreadPoolScheduler.Instance).Subscribe(
    _ => { throw new Exception(); });

EDIT:

Right - so I stopped being lazy and started reading the source code. It turns out the behaviour depends on whether you are on .NET 4.5 or not.

On .NET 4.5 the default behaviour of the task pool is NOT to let unobserved exceptions bring down the process. On .NET 4.0 it does. So Rx follows the platform default and behaves as per other schedulers on .NET 4.0 but silently swallows the exception on .NET 4.5.

So, assuming you are on .NET 4.0, what you are seeing is expected - you have an unobserved exception.

Note also that because .NET 4.5 is an in place upgrade this is a breaking change. It's what Framework is installed on the machine that counts, not what Framework is selected in the project properties.

Source code that's relevant is here - worth a read, the comments are instructive:

http://rx.codeplex.com/SourceControl/latest#Rx.NET/Source/System.Reactive.PlatformServices/Reactive/Concurrency/TaskPoolScheduler.cs

Here are some comments on MSDN about this behaviour change (see the remarks), which also show how to impose .NET 4.0 behaviour on a .NET 4.5 installation:

http://msdn.microsoft.com/en-us/library/jj160346(v=vs.110).aspx

My advice is to not rely on this behaviour but protect your OnNext handlers with exception handlers. To be clear, I mean Subscribers should not throw. The Rx Design guidelines are explicit that calls to OnNext, OnError and OnCompleted should NOT be protected. (See section 6.4 at http://go.microsoft.com/fwlink/?LinkID=205219)

Answer 2

James World thanks for your help. I'm using .Net 4.0 and currently cannot move to 4.5. In my task I need that

public IObservable<IEnumerable<int>>> MultiResponse

would always be available for subscribers. For this I've created the following extension method:

public static class ObservableExtensions
{
    public static IObservable<T> ObserveSafe<T>(this IObservable<T> observable, Action<Exception> onError)
    {
        var subject = new Subject<T>();

        observable.Subscribe(x =>
                             {
                                 try
                                 {
                                     subject.OnNext(x);
                                 }
                                 catch (Exception exception)
                                 {
                                     onError(exception);
                                 }
                             },
                             exception =>
                             {
                                 try
                                 {
                                     subject.OnError(exception);
                                 }
                                 catch (Exception ex)
                                 {
                                     onError(ex);
                                 }
                                 finally
                                 {
                                     subject.Dispose();
                                 }
                             },
                             () =>
                             {
                                 try
                                 {
                                     subject.OnCompleted();
                                 }
                                 catch (Exception ex)
                                 {
                                     onError(ex);
                                 }
                                 finally
                                 {
                                     subject.Dispose();
                                 }
                             });

        return subject;
    }
}