2.x: Observable.repeatWhen fix for onError (#4819)

Author: akarnokd

src/main/java/io/reactivex/Observable.java

@@ -8566,7 +8566,7 @@ public final Observable<T> repeatUntil(BooleanSupplier stop) {
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Observable<T> repeatWhen(final Function<? super Observable<Object>, ? extends ObservableSource<?>> handler) {
         ObjectHelper.requireNonNull(handler, "handler is null");
-        return RxJavaPlugins.onAssembly(new ObservableRedo<T>(this, ObservableInternalHelper.repeatWhenHandler(handler)));
+        return RxJavaPlugins.onAssembly(new ObservableRedo<T>(this, ObservableInternalHelper.repeatWhenHandler(handler), false));
     }
 
     /**
@@ -9219,7 +9219,7 @@ public final Observable<T> retryUntil(final BooleanSupplier stop) {
     public final Observable<T> retryWhen(
             final Function<? super Observable<Throwable>, ? extends ObservableSource<?>> handler) {
         ObjectHelper.requireNonNull(handler, "handler is null");
-        return RxJavaPlugins.onAssembly(new ObservableRedo<T>(this, ObservableInternalHelper.retryWhenHandler(handler)));
+        return RxJavaPlugins.onAssembly(new ObservableRedo<T>(this, ObservableInternalHelper.retryWhenHandler(handler), true));
     }
 
     /**

src/main/java/io/reactivex/internal/operators/observable/ObservableRedo.java

@@ -27,18 +27,22 @@
 public final class ObservableRedo<T> extends AbstractObservableWithUpstream<T, T> {
     final Function<? super Observable<Notification<Object>>, ? extends ObservableSource<?>> manager;
 
+    final boolean retryMode;
+
     public ObservableRedo(ObservableSource<T> source,
-            Function<? super Observable<Notification<Object>>, ? extends ObservableSource<?>> manager) {
+            Function<? super Observable<Notification<Object>>, ? extends ObservableSource<?>> manager,
+                    boolean retryMode) {
         super(source);
         this.manager = manager;
+        this.retryMode = retryMode;
     }
 
     @Override
     public void subscribeActual(Observer<? super T> s) {
 
         Subject<Notification<Object>> subject = BehaviorSubject.<Notification<Object>>create().toSerialized();
 
-        final RedoObserver<T> parent = new RedoObserver<T>(s, subject, source);
+        final RedoObserver<T> parent = new RedoObserver<T>(s, subject, source, retryMode);
 
         ToNotificationObserver<Object> actionObserver = new ToNotificationObserver<Object>(new Consumer<Notification<Object>>() {
             @Override
@@ -73,13 +77,16 @@ static final class RedoObserver<T> extends AtomicBoolean implements Observer<T>
         final ObservableSource<? extends T> source;
         final SequentialDisposable arbiter;
 
+        final boolean retryMode;
+
         final AtomicInteger wip = new AtomicInteger();
 
-        RedoObserver(Observer<? super T> actual, Subject<Notification<Object>> subject, ObservableSource<? extends T> source) {
+        RedoObserver(Observer<? super T> actual, Subject<Notification<Object>> subject, ObservableSource<? extends T> source, boolean retryMode) {
             this.actual = actual;
             this.subject = subject;
             this.source = source;
             this.arbiter = new SequentialDisposable();
+            this.retryMode = retryMode;
             this.lazySet(true);
         }
 
@@ -96,14 +103,22 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (compareAndSet(false, true)) {
-                subject.onNext(Notification.createOnError(t));
+                if (retryMode) {
+                    subject.onNext(Notification.createOnError(t));
+                } else {
+                    subject.onError(t);
+                }
             }
         }
 
         @Override
         public void onComplete() {
             if (compareAndSet(false, true)) {
-                subject.onNext(Notification.createOnComplete());
+                if (retryMode) {
+                    subject.onComplete();
+                } else {
+                    subject.onNext(Notification.createOnComplete());
+                }
             }
         }
 

src/test/java/io/reactivex/internal/operators/flowable/FlowableRepeatTest.java

@@ -329,4 +329,18 @@ public Flowable<Object> apply(Object ignore) throws Exception {
       disposable.dispose();
       assertFalse(subject.hasSubscribers());
     }
+
+    @Test
+    public void testRepeatWhen() {
+        Flowable.error(new TestException())
+        .repeatWhen(new Function<Flowable<Object>, Flowable<Object>>() {
+            @Override
+            public Flowable<Object> apply(Flowable<Object> v) throws Exception {
+                return v.delay(10, TimeUnit.SECONDS);
+            }
+        })
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertFailure(TestException.class);
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableRepeatTest.java

@@ -280,4 +280,18 @@ public ObservableSource<Object> apply(Object ignore) throws Exception {
       disposable.dispose();
       assertFalse(subject.hasObservers());
     }
+
+    @Test
+    public void testRepeatWhen() {
+        Observable.error(new TestException())
+        .repeatWhen(new Function<Observable<Object>, ObservableSource<Object>>() {
+            @Override
+            public ObservableSource<Object> apply(Observable<Object> v) throws Exception {
+                return v.delay(10, TimeUnit.SECONDS);
+            }
+        })
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertFailure(TestException.class);
+    }
 }