Spit IObservable<T> exceptions in a separate IObservable<Exception> and continue normally

user_1139856 0
Viewed 47

I have an hot IObservable<T> which may throw an exception. However, I would like to continue with it. I think I could use Retry operator for that. However, it would be great if I can also listen to any error in IObservable<T> through a separate IObservable<Exception>. Is it possible?

system.reactive
2 Answers

Your case is significantly more simplified in that you have a hot observable.

OnError is a notification outside your value stream, so we could materialize the notifications to retrieve the error. This still causes the tear-down of the stream with an OnCompleted, so you'll need to re-subscribe with Repeat.

var exceptions =
    source
        .Materialize()
        .Where(notif => notif.Kind == NotificationKind.OnError)
        .Select(notif => notif.Exception)
        .Repeat();

Note

If you're using a Subject<T> for your hot observable, you might run into the usual problem of re-subbing a subject. A subject will replay its OnError or OnCompleted notifications for every new observer.

var source = new Subject<int>();            
source.OnNext(1);
source.OnError(new Exception());
source.Subscribe(
        i => Console.WriteLine(i), 
        ex => Console.WriteLine("Still got exception after the throw")
);

In this case your exception stream will go into an infinite re-subscription loop.

user_8021160

The premise of your question violates the observable contract:

An Observable may make zero or more OnNext notifications, each representing a single emitted item, and it may then follow those emission notifications by either an OnCompleted or an OnError notification, but not both. Upon issuing an OnCompleted or OnError notification, it may not thereafter issue any further notifications. (emphasis mine)

In other words, after your hot IObservable<T> throws an exception, the observable is ended. The observable of exceptions that comes out of that has a max count of one.

If you want to support a scenario where you re-start an observable after an exception, you're producing a stream of observables, or IObservable<IObservable<T>>. To work with that, here's a code sample:

var source = new Subject<Subject<int>>();

var exceptionStream = source
    .SelectMany(o => o.Materialize())
    .Where(n => n.Kind == NotificationKind.OnError)
    .Select(n => n.Exception);

var itemStream = source
    .SelectMany(o => o.Materialize())
    .Where(n => n.Kind == NotificationKind.OnNext)
    .Select(n => n.Value);

var items = new List<int>();
var exceptions = new List<Exception>();
itemStream.Subscribe(i => items.Add(i));
exceptionStream.Subscribe(e => exceptions.Add(e));

var currentSubject = new Subject<int>();
source.OnNext(currentSubject);

currentSubject.OnNext(1);
currentSubject.OnNext(2);
currentSubject.OnNext(3);
currentSubject.OnError(new Exception("First error"));

var currentSubject2 = new Subject<int>();
source.OnNext(currentSubject2);

currentSubject2.OnNext(4);
currentSubject2.OnNext(5);
currentSubject2.OnNext(6);
currentSubject2.OnError(new Exception("Second error"));

items.Dump(); //Linqpad
exceptions.Dump(); //Linqpad
user_4156610
Related