2.x: Fix zip not stopping the subscription upon eager error (#6488)

ReactiveX · May 27, 2019 · c415b32 · c415b32
1 parent 4ca7a9b
commit c415b32
Show file tree

Hide file tree

Showing 3 changed files with 64 additions and 0 deletions.
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableZip.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableZip.java
@@ -179,6 +179,7 @@ public void drain() {
                             if (z.done && !delayError) {
                                 Throwable ex = z.error;
                                 if (ex != null) {
+                                    cancelled = true;
                                     cancel();
                                     a.onError(ex);
                                     return;
@@ -224,6 +225,7 @@ boolean checkTerminated(boolean d, boolean empty, Observer<? super R> a, boolean
                 if (delayError) {
                     if (empty) {
                         Throwable e = source.error;
+                        cancelled = true;
                         cancel();
                         if (e != null) {
                             a.onError(e);
@@ -235,11 +237,13 @@ boolean checkTerminated(boolean d, boolean empty, Observer<? super R> a, boolean
                 } else {
                     Throwable e = source.error;
                     if (e != null) {
+                        cancelled = true;
                         cancel();
                         a.onError(e);
                         return true;
                     } else
                     if (empty) {
+                        cancelled = true;
                         cancel();
                         a.onComplete();
                         return true;

diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableZipTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableZipTest.java
@@ -1895,4 +1895,34 @@ public Integer apply(Integer a, Integer b) throws Exception {
 
         ts.assertResult(4);
     }
+
+    @Test
+    public void firstErrorPreventsSecondSubscription() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        List<Flowable<?>> flowableList = new ArrayList<Flowable<?>>();
+        flowableList.add(Flowable.create(new FlowableOnSubscribe<Object>() {
+            @Override
+            public void subscribe(FlowableEmitter<Object> e)
+                    throws Exception { throw new TestException(); }
+        }, BackpressureStrategy.MISSING));
+        flowableList.add(Flowable.create(new FlowableOnSubscribe<Object>() {
+            @Override
+            public void subscribe(FlowableEmitter<Object> e)
+                    throws Exception { counter.getAndIncrement(); }
+        }, BackpressureStrategy.MISSING));
+
+        Flowable.zip(flowableList,
+                new Function<Object[], Object>() {
+                    @Override
+                    public Object apply(Object[] a) throws Exception {
+                        return a;
+                    }
+                })
+        .test()
+        .assertFailure(TestException.class)
+        ;
+
+        assertEquals(0, counter.get());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableZipTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableZipTest.java
@@ -1428,4 +1428,34 @@ public Integer apply(Integer t1, Integer t2) throws Exception {
         ps2.onNext(2);
         to.assertResult(3);
     }
+
+    @Test
+    public void firstErrorPreventsSecondSubscription() {
+        final AtomicInteger counter = new AtomicInteger();
+
+        List<Observable<?>> observableList = new ArrayList<Observable<?>>();
+        observableList.add(Observable.create(new ObservableOnSubscribe<Object>() {
+            @Override
+            public void subscribe(ObservableEmitter<Object> e)
+                    throws Exception { throw new TestException(); }
+        }));
+        observableList.add(Observable.create(new ObservableOnSubscribe<Object>() {
+            @Override
+            public void subscribe(ObservableEmitter<Object> e)
+                    throws Exception { counter.getAndIncrement(); }
+        }));
+
+        Observable.zip(observableList,
+                new Function<Object[], Object>() {
+                    @Override
+                    public Object apply(Object[] a) throws Exception {
+                        return a;
+                    }
+                })
+        .test()
+        .assertFailure(TestException.class)
+        ;
+
+        assertEquals(0, counter.get());
+    }
 }