Correct way to end an observable stream in error

Question

I have several methods which return IObservable. In all cases I set up a query which will cause the returned observable to complete. Generally I've been using the TakeUntil extension method. The observable type I'm using in the TakeUntil contains a flag that tells me if there has been a problem. How could I use this to cause my returned observable to end in error? I would love a TakeUntil overload that allowed the observable to end in error.

Currently I have hacked the method to return a subject which has subscribed to the query observable and also subscribed to other observable that I use in TakeUntil to either call OnCompleted or OnError. I realise this a bad plan but what should I be doing? Any help much appreciated.

Answer 1

Unlike TakeUntil, you have to decide from one of three cases, corresponding to OnNext, OnError and OnCompleteted. And there's already a built-in type which captures it in Notification<T>.

All we need to do is transform concrete notifications into implicit ones - with the Dematerialize operator.

Here's an example stream which throws if the value 10 occurs in the stream, but completes it if any value is >= 9.

        var errorAt10 =
            values.Select(value =>
            {
                if (value == 10)
                    return Notification.CreateOnError<long>(new Exception());

                if (value >= 9)
                    return Notification.CreateOnCompleted<long>();

                return Notification.CreateOnNext(value);
            })
            .Dematerialize();

We could simplify this if we wanted to:

    public static IObservable<T> NotifyAs<T>(this IObservable<T> source, Func<T, NotificationKind> choice, Exception exception = default)
    {
        return source.Select(value =>
        {
            switch (choice(value))
            {
                case NotificationKind.OnError:
                    return Notification.CreateOnError<T>(exception ?? new Exception());
                case NotificationKind.OnCompleted:
                    return Notification.CreateOnCompleted<T>();
                default:
                    return Notification.CreateOnNext(value);
            }

        })
        .Dematerialize();
    }

Now you can rewrite the former example as:

        var errorAt10 =
            values.NotifyAs(value =>
                value == 10 ? NotificationKind.OnError :
                value >= 9 ? NotificationKind.OnCompleted :
                NotificationKind.OnNext
            );

Answer 2

You don't really need to do anything. What you're asking for is already built-in.

If you start with this code:

var source = Observable.Interval(TimeSpan.FromSeconds(1.0));
var ender = new Subject<Unit>();
var query = source.TakeUntil(ender);
query.Subscribe(x => Console.WriteLine(x));

Then you only need to call ender.OnError(new Exception("My exception")); for the source observable to end in error with the exception new Exception("My exception").