2.x: add missing null check to fused Observable.fromCallable

src/main/java/io/reactivex/internal/operators/observable/ObservableFromCallable.java

@@ -54,6 +54,6 @@ public void subscribeActual(Observer<? super T> s) {
 
     @Override
     public T call() throws Exception {
-        return callable.call();
+        return ObjectHelper.requireNonNull(callable.call(), "The callable returned a null value");
     }
 }

src/test/java/io/reactivex/internal/operators/flowable/FlowableFromCallableTest.java

@@ -17,6 +17,7 @@
 package io.reactivex.internal.operators.flowable;
 
 import static org.mockito.Mockito.*;
+import static org.junit.Assert.*;
 
 import java.util.concurrent.*;
 
@@ -26,6 +27,7 @@
 import org.reactivestreams.*;
 
 import io.reactivex.*;
+import io.reactivex.functions.Function;
 import io.reactivex.schedulers.Schedulers;
 import io.reactivex.subscribers.TestSubscriber;
 
@@ -156,4 +158,84 @@ public Object call() throws Exception {
         verify(observer).onError(checkedException);
         verifyNoMoreInteractions(observer);
     }
+
+    @Test
+    public void fusedFlatMapExecution() {
+        final int[] calls = { 0 };
+
+        Flowable.just(1).flatMap(new Function<Integer, Publisher<? extends Object>>() {
+            @Override
+            public Publisher<? extends Object> apply(Integer v)
+                    throws Exception {
+                return Flowable.fromCallable(new Callable<Object>() {
+                    @Override
+                    public Object call() throws Exception {
+                        return ++calls[0];
+                    }
+                });
+            }
+        })
+        .test()
+        .assertResult(1);
+
+        assertEquals(1, calls[0]);
+    }
+
+    @Test
+    public void fusedFlatMapExecutionHidden() {
+        final int[] calls = { 0 };
+
+        Flowable.just(1).hide().flatMap(new Function<Integer, Publisher<? extends Object>>() {
+            @Override
+            public Publisher<? extends Object> apply(Integer v)
+                    throws Exception {
+                return Flowable.fromCallable(new Callable<Object>() {
+                    @Override
+                    public Object call() throws Exception {
+                        return ++calls[0];
+                    }
+                });
+            }
+        })
+        .test()
+        .assertResult(1);
+
+        assertEquals(1, calls[0]);
+    }
+
+    @Test
+    public void fusedFlatMapNull() {
+        Flowable.just(1).flatMap(new Function<Integer, Publisher<? extends Object>>() {
+            @Override
+            public Publisher<? extends Object> apply(Integer v)
+                    throws Exception {
+                return Flowable.fromCallable(new Callable<Object>() {
+                    @Override
+                    public Object call() throws Exception {
+                        return null;
+                    }
+                });
+            }
+        })
+        .test()
+        .assertFailure(NullPointerException.class);
+    }
+
+    @Test
+    public void fusedFlatMapNullHidden() {
+        Flowable.just(1).hide().flatMap(new Function<Integer, Publisher<? extends Object>>() {
+            @Override
+            public Publisher<? extends Object> apply(Integer v)
+                    throws Exception {
+                return Flowable.fromCallable(new Callable<Object>() {
+                    @Override
+                    public Object call() throws Exception {
+                        return null;
+                    }
+                });
+            }
+        })
+        .test()
+        .assertFailure(NullPointerException.class);
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableFromCallableTest.java

@@ -0,0 +1,242 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.reactivex.internal.operators.observable;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+import java.util.concurrent.*;
+
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import io.reactivex.*;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.functions.Function;
+import io.reactivex.observers.TestObserver;
+import io.reactivex.schedulers.Schedulers;
+
+public class ObservableFromCallableTest {
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldNotInvokeFuncUntilSubscription() throws Exception {
+        Callable<Object> func = mock(Callable.class);
+
+        when(func.call()).thenReturn(new Object());
+
+        Observable<Object> fromCallableObservable = Observable.fromCallable(func);
+
+        verifyZeroInteractions(func);
+
+        fromCallableObservable.subscribe();
+
+        verify(func).call();
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldCallOnNextAndOnCompleted() throws Exception {
+        Callable<String> func = mock(Callable.class);
+
+        when(func.call()).thenReturn("test_value");
+
+        Observable<String> fromCallableObservable = Observable.fromCallable(func);
+
+        Observer<Object> observer = TestHelper.mockObserver();
+
+        fromCallableObservable.subscribe(observer);
+
+        verify(observer).onNext("test_value");
+        verify(observer).onComplete();
+        verify(observer, never()).onError(any(Throwable.class));
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldCallOnError() throws Exception {
+        Callable<Object> func = mock(Callable.class);
+
+        Throwable throwable = new IllegalStateException("Test exception");
+        when(func.call()).thenThrow(throwable);
+
+        Observable<Object> fromCallableObservable = Observable.fromCallable(func);
+
+        Observer<Object> observer = TestHelper.mockObserver();
+
+        fromCallableObservable.subscribe(observer);
+
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onComplete();
+        verify(observer).onError(throwable);
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void shouldNotDeliverResultIfSubscriberUnsubscribedBeforeEmission() throws Exception {
+        Callable<String> func = mock(Callable.class);
+
+        final CountDownLatch funcLatch = new CountDownLatch(1);
+        final CountDownLatch observerLatch = new CountDownLatch(1);
+
+        when(func.call()).thenAnswer(new Answer<String>() {
+            @Override
+            public String answer(InvocationOnMock invocation) throws Throwable {
+                observerLatch.countDown();
+
+                try {
+                    funcLatch.await();
+                } catch (InterruptedException e) {
+                    // It's okay, unsubscription causes Thread interruption
+
+                    // Restoring interruption status of the Thread
+                    Thread.currentThread().interrupt();
+                }
+
+                return "should_not_be_delivered";
+            }
+        });
+
+        Observable<String> fromCallableObservable = Observable.fromCallable(func);
+
+        Observer<Object> observer = TestHelper.mockObserver();
+
+        TestObserver<String> outer = new TestObserver<String>(observer);
+
+        fromCallableObservable
+                .subscribeOn(Schedulers.computation())
+                .subscribe(outer);
+
+        // Wait until func will be invoked
+        observerLatch.await();
+
+        // Unsubscribing before emission
+        outer.cancel();
+
+        // Emitting result
+        funcLatch.countDown();
+
+        // func must be invoked
+        verify(func).call();
+
+        // Observer must not be notified at all
+        verify(observer).onSubscribe(any(Disposable.class));
+        verifyNoMoreInteractions(observer);
+    }
+
+    @Test
+    public void shouldAllowToThrowCheckedException() {
+        final Exception checkedException = new Exception("test exception");
+
+        Observable<Object> fromCallableObservable = Observable.fromCallable(new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+                throw checkedException;
+            }
+        });
+
+        Observer<Object> observer = TestHelper.mockObserver();
+
+        fromCallableObservable.subscribe(observer);
+
+        verify(observer).onSubscribe(any(Disposable.class));
+        verify(observer).onError(checkedException);
+        verifyNoMoreInteractions(observer);
+    }
+
+    @Test
+    public void fusedFlatMapExecution() {
+        final int[] calls = { 0 };
+
+        Observable.just(1).flatMap(new Function<Integer, ObservableSource<? extends Object>>() {
+            @Override
+            public ObservableSource<? extends Object> apply(Integer v)
+                    throws Exception {
+                return Observable.fromCallable(new Callable<Object>() {
+                    @Override
+                    public Object call() throws Exception {
+                        return ++calls[0];
+                    }
+                });
+            }
+        })
+        .test()
+        .assertResult(1);
+
+        assertEquals(1, calls[0]);
+    }
+
+    @Test
+    public void fusedFlatMapExecutionHidden() {
+        final int[] calls = { 0 };
+
+        Observable.just(1).hide().flatMap(new Function<Integer, ObservableSource<? extends Object>>() {
+            @Override
+            public ObservableSource<? extends Object> apply(Integer v)
+                    throws Exception {
+                return Observable.fromCallable(new Callable<Object>() {
+                    @Override
+                    public Object call() throws Exception {
+                        return ++calls[0];
+                    }
+                });
+            }
+        })
+        .test()
+        .assertResult(1);
+
+        assertEquals(1, calls[0]);
+    }
+
+    @Test
+    public void fusedFlatMapNull() {
+        Observable.just(1).flatMap(new Function<Integer, ObservableSource<? extends Object>>() {
+            @Override
+            public ObservableSource<? extends Object> apply(Integer v)
+                    throws Exception {
+                return Observable.fromCallable(new Callable<Object>() {
+                    @Override
+                    public Object call() throws Exception {
+                        return null;
+                    }
+                });
+            }
+        })
+        .test()
+        .assertFailure(NullPointerException.class);
+    }
+
+    @Test
+    public void fusedFlatMapNullHidden() {
+        Observable.just(1).hide().flatMap(new Function<Integer, ObservableSource<? extends Object>>() {
+            @Override
+            public ObservableSource<? extends Object> apply(Integer v)
+                    throws Exception {
+                return Observable.fromCallable(new Callable<Object>() {
+                    @Override
+                    public Object call() throws Exception {
+                        return null;
+                    }
+                });
+            }
+        })
+        .test()
+        .assertFailure(NullPointerException.class);
+    }
+}