diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFromCallable.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFromCallable.java index 5a773dffc4..6dcb226daa 100644 --- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableFromCallable.java +++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableFromCallable.java @@ -21,6 +21,7 @@ import io.reactivex.exceptions.Exceptions; import io.reactivex.internal.functions.ObjectHelper; import io.reactivex.internal.subscriptions.DeferredScalarSubscription; +import io.reactivex.plugins.RxJavaPlugins; public final class FlowableFromCallable extends Flowable implements Callable { final Callable callable; @@ -38,7 +39,11 @@ public void subscribeActual(Subscriber s) { t = ObjectHelper.requireNonNull(callable.call(), "The callable returned a null value"); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); - s.onError(ex); + if (deferred.isCancelled()) { + RxJavaPlugins.onError(ex); + } else { + s.onError(ex); + } return; } diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableFromCallableTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableFromCallableTest.java index f3f5e00fa5..f3239001ee 100644 --- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableFromCallableTest.java +++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableFromCallableTest.java @@ -16,9 +16,11 @@ package io.reactivex.internal.operators.flowable; +import static org.junit.Assert.assertEquals; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.*; -import static org.junit.Assert.*; +import java.util.List; import java.util.concurrent.*; import org.junit.Test; @@ -27,7 +29,9 @@ import org.reactivestreams.*; import io.reactivex.*; +import io.reactivex.exceptions.TestException; import io.reactivex.functions.Function; +import io.reactivex.plugins.RxJavaPlugins; import io.reactivex.schedulers.Schedulers; import io.reactivex.subscribers.TestSubscriber; @@ -238,4 +242,27 @@ public Object call() throws Exception { .test() .assertFailure(NullPointerException.class); } + + @Test(timeout = 5000) + public void undeliverableUponCancellation() throws Exception { + List errors = TestHelper.trackPluginErrors(); + try { + final TestSubscriber ts = new TestSubscriber(); + + Flowable.fromCallable(new Callable() { + @Override + public Integer call() throws Exception { + ts.cancel(); + throw new TestException(); + } + }) + .subscribe(ts); + + ts.assertEmpty(); + + TestHelper.assertUndeliverable(errors, 0, TestException.class); + } finally { + RxJavaPlugins.reset(); + } + } }