From 8db356994aa1ffb7553261e0404527125279c391 Mon Sep 17 00:00:00 2001
From: David Karnok <akarnokd@gmail.com>
Date: Wed, 14 Aug 2019 12:12:41 +0200
Subject: [PATCH] 2.x: Fix switchMap incorrect sync-fusion & error management
 (#6618)

---
 .../operators/flowable/FlowableSwitchMap.java |  9 ++++-
 .../observable/ObservableSwitchMap.java       |  1 +
 .../flowable/FlowableSwitchTest.java          | 28 +++++++++++++++
 .../observable/ObservableSwitchTest.java      | 35 +++++++++++++++++--
 4 files changed, 70 insertions(+), 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
index 9730bd38b2..d3832d4edc 100644
--- 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
index 8c5aa371dc..4e97ea405d 100644
--- 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
index c1d8c4134c..b4880d4ff6 100644
--- 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
index 7e3d4dc505..100052f028 100644
--- a/src/test/java/io/reactivex/internal/operators/observable/ObservableSwitchTest.java
+++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableSwitchTest.java
@@ -14,9 +14,10 @@
 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.*;
 
@@ -24,6 +25,8 @@
 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);
+    }
 }