From 53a46bd5d84114e67832b9dfe30c41e62fbd8ded Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?D=C3=A1vid=20Karnok?= <akarnokd@gmail.com>
Date: Mon, 7 Nov 2016 19:22:21 +0100
Subject: [PATCH] 2.x: Observable.repeatWhen fix for onError

---
 src/main/java/io/reactivex/Observable.java    |  4 +--
 .../operators/observable/ObservableRedo.java  | 25 +++++++++++++++----
 .../flowable/FlowableRepeatTest.java          | 14 +++++++++++
 .../observable/ObservableRepeatTest.java      | 14 +++++++++++
 4 files changed, 50 insertions(+), 7 deletions(-)

diff --git a/src/main/java/io/reactivex/Observable.java b/src/main/java/io/reactivex/Observable.java
index a9f71a64a7..5757b0a999 100644
--- a/src/main/java/io/reactivex/Observable.java
+++ b/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));
     }
 
     /**
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableRedo.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableRedo.java
index 59c8ebf05f..14b1cb8413 100644
--- a/src/main/java/io/reactivex/internal/operators/observable/ObservableRedo.java
+++ b/src/main/java/io/reactivex/internal/operators/observable/ObservableRedo.java
@@ -27,10 +27,14 @@
 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
@@ -38,7 +42,7 @@ 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());
+                }
             }
         }
 
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableRepeatTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableRepeatTest.java
index 96d577eb0a..40989b5363 100644
--- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableRepeatTest.java
+++ b/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);
+    }
 }
\ No newline at end of file
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableRepeatTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableRepeatTest.java
index acf87c8992..64a5e68adf 100644
--- a/src/test/java/io/reactivex/internal/operators/observable/ObservableRepeatTest.java
+++ b/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);
+    }
 }