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 1a337c1066..7deccd3e12 100644 --- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchMap.java +++ b/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); + } + } } } 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 fb4c762518..724817c421 100644 --- a/src/main/java/io/reactivex/internal/operators/observable/ObservableSwitchMap.java +++ b/src/main/java/io/reactivex/internal/operators/observable/ObservableSwitchMap.java @@ -316,6 +316,7 @@ void innerError(SwitchMapInnerObserver 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 4415f4c9e0..beade2f3ba 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 Publisher apply(Integer v) throws Throwable { RxJavaPlugins.reset(); } } + + @Test + public void switchMapFusedIterable() { + Flowable.range(1, 2) + .switchMap(new Function>() { + @Override + public Publisher 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>() { + @Override + public Publisher apply(Integer v) + throws Throwable { + 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 b947f49915..eb72e2725d 100644 --- a/src/test/java/io/reactivex/internal/operators/observable/ObservableSwitchTest.java +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableSwitchTest.java @@ -17,7 +17,7 @@ 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.*; @@ -25,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.*; @@ -1226,4 +1228,32 @@ public Observable apply(Integer v) throws Throwable { RxJavaPlugins.reset(); } } + + @Test + public void switchMapFusedIterable() { + Observable.range(1, 2) + .switchMap(new Function>() { + @Override + public Observable 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>() { + @Override + public Observable apply(Integer v) + throws Throwable { + return Observable.fromIterable(Arrays.asList(v * 10)).hide(); + } + }) + .test() + .assertResult(10, 20); + } }