From 547868fc075b727a0d75664c390485130c12b23c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?D=C3=A1vid=20Karnok?= Date: Tue, 7 Aug 2018 21:52:36 +0200 Subject: [PATCH] 2.x: Fix boundary fusion of concatMap and publish operator --- .../operators/flowable/FlowableConcatMap.java | 2 +- .../operators/flowable/FlowablePublish.java | 16 +-- .../flowable/FlowableConcatMapTest.java | 104 ++++++++++++++++++ .../flowable/FlowablePublishTest.java | 51 +++++++++ 4 files changed, 164 insertions(+), 9 deletions(-) diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMap.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMap.java index 27016a1dd8..4dc60faa5a 100644 --- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMap.java +++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableConcatMap.java @@ -113,7 +113,7 @@ public final void onSubscribe(Subscription s) { if (s instanceof QueueSubscription) { @SuppressWarnings("unchecked") QueueSubscription f = (QueueSubscription)s; - int m = f.requestFusion(QueueSubscription.ANY); + int m = f.requestFusion(QueueSubscription.ANY | QueueSubscription.BOUNDARY); if (m == QueueSubscription.SYNC) { sourceMode = m; queue = f; diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublish.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublish.java index 9bc0d63b65..6122974695 100644 --- a/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublish.java +++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublish.java @@ -155,7 +155,7 @@ static final class PublishSubscriber */ final AtomicBoolean shouldConnect; - final AtomicReference s = new AtomicReference(); + final AtomicReference upstream = new AtomicReference(); /** Contains either an onComplete or an onError token from upstream. */ volatile Object terminalEvent; @@ -180,7 +180,7 @@ public void dispose() { InnerSubscriber[] ps = subscribers.getAndSet(TERMINATED); if (ps != TERMINATED) { current.compareAndSet(PublishSubscriber.this, null); - SubscriptionHelper.cancel(s); + SubscriptionHelper.cancel(upstream); } } } @@ -192,12 +192,12 @@ public boolean isDisposed() { @Override public void onSubscribe(Subscription s) { - if (SubscriptionHelper.setOnce(this.s, s)) { + if (SubscriptionHelper.setOnce(this.upstream, s)) { if (s instanceof QueueSubscription) { @SuppressWarnings("unchecked") QueueSubscription qs = (QueueSubscription) s; - int m = qs.requestFusion(QueueSubscription.ANY); + int m = qs.requestFusion(QueueSubscription.ANY | QueueSubscription.BOUNDARY); if (m == QueueSubscription.SYNC) { sourceMode = m; queue = qs; @@ -482,7 +482,7 @@ void dispatch() { v = q.poll(); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); - s.get().cancel(); + upstream.get().cancel(); term = NotificationLite.error(ex); terminalEvent = term; v = null; @@ -493,7 +493,7 @@ void dispatch() { } // otherwise, just ask for a new value if (sourceMode != QueueSubscription.SYNC) { - s.get().request(1); + upstream.get().request(1); } // and retry emitting to potential new child subscribers continue; @@ -510,7 +510,7 @@ void dispatch() { v = q.poll(); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); - s.get().cancel(); + upstream.get().cancel(); term = NotificationLite.error(ex); terminalEvent = term; v = null; @@ -562,7 +562,7 @@ void dispatch() { // if we did emit at least one element, request more to replenish the queue if (d > 0) { if (sourceMode != QueueSubscription.SYNC) { - s.get().request(d); + upstream.get().request(d); } } // if we have requests but not an empty queue after emission diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatMapTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatMapTest.java index c1ad560478..ac5c573910 100644 --- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatMapTest.java +++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatMapTest.java @@ -13,9 +13,16 @@ package io.reactivex.internal.operators.flowable; +import java.util.concurrent.TimeUnit; + import org.junit.Test; +import org.reactivestreams.Publisher; +import io.reactivex.*; +import io.reactivex.exceptions.TestException; +import io.reactivex.functions.Function; import io.reactivex.internal.operators.flowable.FlowableConcatMap.WeakScalarSubscription; +import io.reactivex.schedulers.Schedulers; import io.reactivex.subscribers.TestSubscriber; public class FlowableConcatMapTest { @@ -39,4 +46,101 @@ public void weakSubscriptionRequest() { ts.assertResult(1); } + @Test + public void boundaryFusion() { + Flowable.range(1, 10000) + .observeOn(Schedulers.single()) + .map(new Function() { + @Override + public String apply(Integer t) throws Exception { + String name = Thread.currentThread().getName(); + if (name.contains("RxSingleScheduler")) { + return "RxSingleScheduler"; + } + return name; + } + }) + .concatMap(new Function>() { + @Override + public Publisher apply(String v) + throws Exception { + return Flowable.just(v); + } + }) + .observeOn(Schedulers.computation()) + .distinct() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult("RxSingleScheduler"); + } + + @Test + public void boundaryFusionDelayError() { + Flowable.range(1, 10000) + .observeOn(Schedulers.single()) + .map(new Function() { + @Override + public String apply(Integer t) throws Exception { + String name = Thread.currentThread().getName(); + if (name.contains("RxSingleScheduler")) { + return "RxSingleScheduler"; + } + return name; + } + }) + .concatMapDelayError(new Function>() { + @Override + public Publisher apply(String v) + throws Exception { + return Flowable.just(v); + } + }) + .observeOn(Schedulers.computation()) + .distinct() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult("RxSingleScheduler"); + } + + @Test + public void pollThrows() { + Flowable.just(1) + .map(new Function() { + @Override + public Integer apply(Integer v) throws Exception { + throw new TestException(); + } + }) + .compose(TestHelper.flowableStripBoundary()) + .concatMap(new Function>() { + @Override + public Publisher apply(Integer v) + throws Exception { + return Flowable.just(v); + } + }) + .test() + .assertFailure(TestException.class); + } + + @Test + public void pollThrowsDelayError() { + Flowable.just(1) + .map(new Function() { + @Override + public Integer apply(Integer v) throws Exception { + throw new TestException(); + } + }) + .compose(TestHelper.flowableStripBoundary()) + .concatMapDelayError(new Function>() { + @Override + public Publisher apply(Integer v) + throws Exception { + return Flowable.just(v); + } + }) + .test() + .assertFailure(TestException.class); + } } diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishTest.java index bf8b269196..c18cbc928b 100644 --- a/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishTest.java +++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowablePublishTest.java @@ -824,12 +824,36 @@ public Object apply(Integer v) throws Exception { throw new TestException(); } }) + .compose(TestHelper.flowableStripBoundary()) .publish() .autoConnect() .test() .assertFailure(TestException.class); } + @Test + public void pollThrowsNoSubscribers() { + ConnectableFlowable cf = Flowable.just(1, 2) + .map(new Function() { + @Override + public Integer apply(Integer v) throws Exception { + if (v == 2) { + throw new TestException(); + } + return v; + } + }) + .compose(TestHelper.flowableStripBoundary()) + .publish(); + + TestSubscriber ts = cf.take(1) + .test(); + + cf.connect(); + + ts.assertResult(1); + } + @Test public void dryRunCrash() { List errors = TestHelper.trackPluginErrors(); @@ -1316,4 +1340,31 @@ public void onComplete() { ts1.assertEmpty(); ts2.assertValuesOnly(1); } + + @Test + public void boundaryFusion() { + Flowable.range(1, 10000) + .observeOn(Schedulers.single()) + .map(new Function() { + @Override + public String apply(Integer t) throws Exception { + String name = Thread.currentThread().getName(); + if (name.contains("RxSingleScheduler")) { + return "RxSingleScheduler"; + } + return name; + } + }) + .share() + .observeOn(Schedulers.computation()) + .distinct() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult("RxSingleScheduler"); + } + + @Test + public void badRequest() { + TestHelper.assertBadRequestReported(Flowable.range(1, 5).publish()); + } }