2.x: Make Flowable.fromCallable consitent with the other fromCallables (#6158)

2.x: Make Flowable.fromCallable consistent with the other fromCallables

Author: akarnokd

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<T> extends Flowable<T> implements Callable<T> {
     final Callable<? extends T> callable;
@@ -38,7 +39,11 @@ public void subscribeActual(Subscriber<? super T> 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;
         }
 

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<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            final TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+
+            Flowable.fromCallable(new Callable<Integer>() {
+                @Override
+                public Integer call() throws Exception {
+                    ts.cancel();
+                    throw new TestException();
+                }
+            })
+            .subscribe(ts);
+
+            ts.assertEmpty();
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class);
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
 }