onErrorResumeNext, refCount and retry

[david-hoze]

Hi, I noticed some behavior I don't fully understand.

When I run this:

Observable<String> interval =
        Observable.interval(1,TimeUnit.SECONDS)
                .doOnSubscribe(() -> System.out.println("subscribe interval"))
                .doOnUnsubscribe(() -> System.out.println("unsubscribe interval"))
                .flatMap(onClickEvent -> Observable.defer(() -> Observable.<String>error(new Exception("Some exception"))))
                .onErrorResumeNext(Observable::error)
                .publish()
                .refCount()
                .doOnSubscribe(() -> System.out.println("subscribe refCount"))
                .doOnUnsubscribe(() -> System.out.println("unsubscribe refCount"));

interval
        .doOnError(throwable -> System.out.println("subscriber 1 error: " + throwable.getMessage()))
        .retry()
        .subscribe(s -> System.out.println("subscriber 1: " + s));

I get:

subscribe refCount
subscribe interval
unsubscribe interval
subscriber 1 error: Some exception
unsubscribe refCount
subscribe refCount
subscribe interval
unsubscribe interval
subscriber 1 error: Some exception
unsubscribe refCount
...

The three dots indicate that it continues infinitely.
Now when I add a second subscriber:

interval
        .doOnError(throwable -> System.out.println("subscriber 2 error: " + throwable.getMessage()))
        .retry()
        .subscribe(s -> System.out.println("subscriber 2: " + s));

I get:

subscribe refCount
subscribe interval
subscribe refCount
unsubscribe interval
subscriber 1 error: Some exception
unsubscribe refCount
subscribe refCount
subscriber 2 error: Some exception
unsubscribe refCount
subscribe refCount

and that's it, it doesn't retry. More specifically, it unsubscribes from interval and does not resubscribe. When I remove the onErrorResumeNext it's working fine like this:

subscribe refCount
subscribe interval
subscribe refCount
subscriber 1 error: Some exception
unsubscribe refCount
subscribe refCount
subscriber 2 error: Some exception
unsubscribe refCount
subscribe refCount
subscriber 1 error: Some exception
unsubscribe refCount
subscribe refCount
subscriber 2 error: Some exception
unsubscribe refCount
subscribe refCount
...

Where the three dots indicate it goes on infinitely. As you notice it doesn't unsubscribe from interval. I'm using onErrorResumeNext to "map" the error (in this example it's a trivial mapping), maybe I shouldn't use it like this. Still, I don't understand what's going on..

[akarnokd]

Hi. This is most likely the manifestation of the publish() bug. Could you check if #2552 works for you?

[david-hoze]

Checked it, exactly the same behavior.

[akarnokd]

The problem happens because retry only retries the refCount which never unsubscribes from the upstream due to two subscribers (they both unsubscribe and resubscribe immediately thus the refCount never reaches zero). The upstream keeps sending timer events which the turn into onError events. onErrorResumeNext stops accepting events after the first error and unsubscribes from the upstream but because refCount never re-subscribes, the whole process stops.

[akarnokd]

I think I fixed the problem. Your example above works with it. Could you also check if #2567 works for you?

[david-hoze]

I think it fixes it!
I'm getting now for (the second example with two subscribers and onErrorResumeNext):

Observable<String> interval =
        Observable.interval(1,TimeUnit.SECONDS)
                .doOnSubscribe(() -> System.out.println("subscribe interval"))
                .doOnUnsubscribe(() -> System.out.println("unsubscribe interval"))
                .flatMap(onClickEvent -> Observable.defer(() -> Observable.<String>error(new Exception("Some exception"))))
                .onErrorResumeNext(Observable::error)
                .publish()
                .refCount()
                .doOnSubscribe(() -> System.out.println("subscribe refCount"))
                .doOnUnsubscribe(() -> System.out.println("unsubscribe refCount"));

interval
        .doOnError(throwable -> System.out.println("subscriber 1 error: " + throwable.getMessage()))
        .retry()
        .subscribe(s -> System.out.println("subscriber 1: " + s));

interval
        .doOnError(throwable -> System.out.println("subscriber 2 error: " + throwable.getMessage()))
        .retry()
        .subscribe(s -> System.out.println("subscriber 2: " + s));

This result:

subscribe refCount
subscribe interval
subscribe refCount
unsubscribe interval
subscriber 1 error: Some exception
unsubscribe refCount
subscribe refCount
subscribe interval
subscriber 2 error: Some exception
unsubscribe refCount
subscribe refCount
unsubscribe interval
subscriber 1 error: Some exception
unsubscribe refCount
subscribe refCount
subscribe interval
subscriber 2 error: Some exception
unsubscribe refCount
subscribe refCount
unsubscribe interval
...

Three dots say it continues infinitely. Notice that it does resubscribe to interval now.

[akarnokd]

Thanks for confirming. I'm still waiting for @davidmoten's review before merging in the fix.

[david-hoze]

No problem

[akarnokd]

The fix is available in 1.0.5. Thanks for reporting.

[david-hoze]

Great