Observable continues to retry subscription after catching error-CodePudding

The below is code I encountered from a project I am working on, simpified as much as I can.

The ViewModel listens on a command (button press) which triggers _subject.OnNext(). It subscribes to _subject as below. There is an extension method, one rror that catches exceptions.

What is happening is that the error is caught as expected but then the observable returned from GetObservable keeps retrying. So I get output like the below non stop. I am not understanding how one rror is working here. Does Catch automatically resubscribe when returning a source?

Output:

SUBSCRIBED
System.Exception: ERROR
SUBSCRIBED
System.Exception: ERROR
SUBSCRIBED
System.Exception: ERROR

Code:

public ViewModel()
{
    _subject.Subscribe(_ => Save());
}

private void Save()
{
    GetObservable().OnError(ex => Debug.WriteLine(ex)).Subscribe();
}

private IObservable<Unit> GetObservable()
{
    return Observable.Create<Unit>(observer =>
    {
        Debug.WriteLine("SUBSCRIBED");
        observer.OnNext(Unit.Default);
        observer.OnError(new Exception("ERROR"));
        observer.OnCompleted();
        return Disposable.Empty;
    }).Timeout(TimeSpan.FromSeconds(60));
}

public static class ObservableExtensions
{
    public static IObservable<T> one rror<T>(this IObservable<T> source, Action<Exception> action)
    {
        IObservable<T> observable = null;
        observable = source.Catch<T, Exception>(ex =>
          {
              action(ex);
              return observable;
          });
        return observable;
    }
}

CodePudding user response：

The Catch() operator returns an observable, which is like "concatenating" the observable it is called on and the observable which is defined in the provided Func<TException, IObservable<TSource>> expression in case there was an exception. Check the marble diagram of the Catch operator from https://reactivex.io/documentation/operators/catch.html:

As you see the result is "one" observable.

So in your case you get:

OnNext(Unit.Default)

and due to the exception the observable you are currently building. So basically you start from the top again. And this means you observable generates:

OnNext(Unit.Default)
OnNext(Unit.Default)
OnNext(Unit.Default)
OnNext(Unit.Default)
OnNext(Unit.Default)
...