2.x: terminating repeatWhen

[ychescale9]

In 2.X repeatWhen can't seem to be terminated (onError) by throwing an exception from the observable:

private int count = 0;
private Observable<Integer> getTestObservable() {
    return Observable.fromCallable(new Callable<Integer>() {
        @Override
        public Integer call() throws Exception {
            count++;
            if (count == 5) {
                throw new Exception("Some exception");
            }
            return count;
        }
    });
}

@Test
public void testRepeatWhen() {
    getTestObservable()
            .repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
                @Override
                public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception {
                    return objectObservable.delay(1, TimeUnit.SECONDS);
                }
            })
            .subscribe(new Observer<Integer>() {
                @Override
                public void onSubscribe(Disposable d) {
                    System.out.println("onSubscribe");
                }

                @Override
                public void onNext(Integer value) {
                    System.out.println("onNext: " + value);
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println("onError");
                }

                @Override
                public void onComplete() {
                    System.out.println("onComplete");
                }
            });
}

Output:

onSubscribe
onNext: 1
onNext: 2
onNext: 3
onNext: 4
onNext: 6
onNext: 7
...

When the exception is thrown, neither onNext nor onError is called on the subscription, then repeatWhen continues.

This behavior is different to the equivalence in 1.x.

How can repeatWhen be terminated in 2.x?

Thanks.

[lucas34]

Use TakeUntil() or takeWhile() http://reactivex.io/documentation/operators/takeuntil.html

[ychescale9]

Thanks but TakeUntil() and takeWhile() are different operators for different use cases. I'm wondering why/how repeatWhen behaves differently in 2.x.

[akarnokd]

Thanks for reporting. This is a bug with Observable.repeatWhen (Flowable.repeatWhen works). I'll post a fix soon.

[akarnokd]

See #4819.

[akarnokd]

Closing via #4819