3.x: Fix switchMap incorrect sync-fusion & error management (#6616)

Author: akarnokd

src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchMap.java

@@ -311,7 +311,7 @@ void drain() {
                             if (r != Long.MAX_VALUE) {
                                 requested.addAndGet(-e);
                             }
-                            inner.get().request(e);
+                            inner.request(e);
                         }
                     }
 
@@ -395,6 +395,7 @@ public void onError(Throwable t) {
             if (index == p.unique && p.errors.addThrowable(t)) {
                 if (!p.delayErrors) {
                     p.upstream.cancel();
+                    p.done = true;
                 }
                 done = true;
                 p.drain();
@@ -415,5 +416,11 @@ public void onComplete() {
         public void cancel() {
             SubscriptionHelper.cancel(this);
         }
+
+        public void request(long n) {
+            if (fusionMode != QueueSubscription.SYNC) {
+                get().request(n);
+            }
+        }
     }
 }

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

@@ -316,6 +316,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();

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

@@ -1201,4 +1201,32 @@ public Publisher<Integer> apply(Integer v) throws Throwable {
             RxJavaPlugins.reset();
         }
     }
+
+    @Test
+    public void switchMapFusedIterable() {
+        Flowable.range(1, 2)
+        .switchMap(new Function<Integer, Publisher<Integer>>() {
+            @Override
+            public Publisher<Integer> apply(Integer v)
+                    throws Throwable {
+                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 Throwable {
+                return Flowable.fromIterable(Arrays.asList(v * 10)).hide();
+            }
+        })
+        .test()
+        .assertResult(10, 20);
+    }
 }

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

@@ -17,14 +17,16 @@
 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.*;
@@ -1226,4 +1228,32 @@ public Observable<Integer> apply(Integer v) throws Throwable {
             RxJavaPlugins.reset();
         }
     }
+
+    @Test
+    public void switchMapFusedIterable() {
+        Observable.range(1, 2)
+        .switchMap(new Function<Integer, Observable<Integer>>() {
+            @Override
+            public Observable<Integer> apply(Integer v)
+                    throws Throwable {
+                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 Throwable {
+                return Observable.fromIterable(Arrays.asList(v * 10)).hide();
+            }
+        })
+        .test()
+        .assertResult(10, 20);
+    }
 }