2.x: Fix switchMap incorrect sync-fusion & error management (#6618)

ReactiveX · Aug 14, 2019 · 8db3569 · 8db3569
1 parent 17a8eef
commit 8db3569
Show file tree

Hide file tree

Showing 4 changed files with 70 additions and 3 deletions.
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchMap.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchMap.java
@@ -314,7 +314,7 @@ void drain() {
                             if (r != Long.MAX_VALUE) {
                                 requested.addAndGet(-e);
                             }
-                            inner.get().request(e);
+                            inner.request(e);
                         }
                     }
 
@@ -398,6 +398,7 @@ public void onError(Throwable t) {
             if (index == p.unique && p.error.addThrowable(t)) {
                 if (!p.delayErrors) {
                     p.upstream.cancel();
+                    p.done = true;
                 }
                 done = true;
                 p.drain();
@@ -418,5 +419,11 @@ public void onComplete() {
         public void cancel() {
             SubscriptionHelper.cancel(this);
         }
+
+        public void request(long n) {
+            if (fusionMode != QueueSubscription.SYNC) {
+                get().request(n);
+            }
+        }
     }
 }
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableSwitchMap.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableSwitchMap.java
@@ -314,6 +314,7 @@ void innerError(SwitchMapInnerObserver<T, R> inner, Throwable ex) {
             if (inner.index == unique && errors.addThrowable(ex)) {
                 if (!delayErrors) {
                     upstream.dispose();
+                    done = true;
                 }
                 inner.done = true;
                 drain();

diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableSwitchTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableSwitchTest.java
@@ -1201,4 +1201,32 @@ public Object apply(Integer w) throws Exception {
         .assertNoErrors()
         .assertComplete();
     }
+
+    @Test
+    public void switchMapFusedIterable() {
+        Flowable.range(1, 2)
+        .switchMap(new Function<Integer, Publisher<Integer>>() {
+            @Override
+            public Publisher<Integer> apply(Integer v)
+                    throws Exception {
+                return Flowable.fromIterable(Arrays.asList(v * 10));
+            }
+        })
+        .test()
+        .assertResult(10, 20);
+    }
+
+    @Test
+    public void switchMapHiddenIterable() {
+        Flowable.range(1, 2)
+        .switchMap(new Function<Integer, Publisher<Integer>>() {
+            @Override
+            public Publisher<Integer> apply(Integer v)
+                    throws Exception {
+                return Flowable.fromIterable(Arrays.asList(v * 10)).hide();
+            }
+        })
+        .test()
+        .assertResult(10, 20);
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableSwitchTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableSwitchTest.java
@@ -14,16 +14,19 @@
 package io.reactivex.internal.operators.observable;
 
 import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
-import java.util.List;
+import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.*;
 
 import org.junit.*;
 import org.mockito.InOrder;
 
 import io.reactivex.*;
+import io.reactivex.Observable;
+import io.reactivex.Observer;
 import io.reactivex.disposables.*;
 import io.reactivex.exceptions.*;
 import io.reactivex.functions.*;
@@ -33,7 +36,7 @@
 import io.reactivex.observers.TestObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.schedulers.*;
-import io.reactivex.subjects.*;
+import io.reactivex.subjects.PublishSubject;
 
 public class ObservableSwitchTest {
 
@@ -1191,4 +1194,32 @@ public Object apply(Integer w) throws Exception {
         .assertNoErrors()
         .assertComplete();
     }
+
+    @Test
+    public void switchMapFusedIterable() {
+        Observable.range(1, 2)
+        .switchMap(new Function<Integer, Observable<Integer>>() {
+            @Override
+            public Observable<Integer> apply(Integer v)
+                    throws Exception {
+                return Observable.fromIterable(Arrays.asList(v * 10));
+            }
+        })
+        .test()
+        .assertResult(10, 20);
+    }
+
+    @Test
+    public void switchMapHiddenIterable() {
+        Observable.range(1, 2)
+        .switchMap(new Function<Integer, Observable<Integer>>() {
+            @Override
+            public Observable<Integer> apply(Integer v)
+                    throws Exception {
+                return Observable.fromIterable(Arrays.asList(v * 10)).hide();
+            }
+        })
+        .test()
+        .assertResult(10, 20);
+    }
 }