2.x: Fix Flowable.flatMap not canceling the inner sources on outer er…

…ror (#6827)
ReactiveX · Jan 9, 2020 · 0b3b300 · 0b3b300
1 parent ea2c796
commit 0b3b300
Show file tree

Hide file tree

Showing 3 changed files with 73 additions and 0 deletions.
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMap.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMap.java
@@ -322,6 +322,11 @@ public void onError(Throwable t) {
             }
             if (errs.addThrowable(t)) {
                 done = true;
+                if (!delayErrors) {
+                    for (InnerSubscriber<?, ?> a : subscribers.getAndSet(CANCELLED)) {
+                        a.dispose();
+                    }
+                }
                 drain();
             } else {
                 RxJavaPlugins.onError(t);

diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableFlatMapTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableFlatMapTest.java
@@ -1123,4 +1123,38 @@ public void accept(Integer v) throws Exception {
         assertFalse(pp3.hasSubscribers());
         assertFalse(pp4.hasSubscribers());
     }
+
+    @Test
+    public void mainErrorsInnerCancelled() {
+        PublishProcessor<Integer> pp1 = PublishProcessor.create();
+        PublishProcessor<Integer> pp2 = PublishProcessor.create();
+
+        pp1
+        .flatMap(Functions.justFunction(pp2))
+        .test();
+
+        pp1.onNext(1);
+        assertTrue("No subscribers?", pp2.hasSubscribers());
+
+        pp1.onError(new TestException());
+
+        assertFalse("Has subscribers?", pp2.hasSubscribers());
+    }
+
+    @Test
+    public void innerErrorsMainCancelled() {
+        PublishProcessor<Integer> pp1 = PublishProcessor.create();
+        PublishProcessor<Integer> pp2 = PublishProcessor.create();
+
+        pp1
+        .flatMap(Functions.justFunction(pp2))
+        .test();
+
+        pp1.onNext(1);
+        assertTrue("No subscribers?", pp2.hasSubscribers());
+
+        pp2.onError(new TestException());
+
+        assertFalse("Has subscribers?", pp1.hasSubscribers());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableFlatMapTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableFlatMapTest.java
@@ -1084,4 +1084,38 @@ public void accept(Integer v) throws Exception {
         assertFalse(ps3.hasObservers());
         assertFalse(ps4.hasObservers());
     }
+
+    @Test
+    public void mainErrorsInnerCancelled() {
+        PublishSubject<Integer> ps1 = PublishSubject.create();
+        PublishSubject<Integer> ps2 = PublishSubject.create();
+
+        ps1
+        .flatMap(Functions.justFunction(ps2))
+        .test();
+
+        ps1.onNext(1);
+        assertTrue("No subscribers?", ps2.hasObservers());
+
+        ps1.onError(new TestException());
+
+        assertFalse("Has subscribers?", ps2.hasObservers());
+    }
+
+    @Test
+    public void innerErrorsMainCancelled() {
+        PublishSubject<Integer> ps1 = PublishSubject.create();
+        PublishSubject<Integer> ps2 = PublishSubject.create();
+
+        ps1
+        .flatMap(Functions.justFunction(ps2))
+        .test();
+
+        ps1.onNext(1);
+        assertTrue("No subscribers?", ps2.hasObservers());
+
+        ps2.onError(new TestException());
+
+        assertFalse("Has subscribers?", ps1.hasObservers());
+    }
 }