ObservableFromCallable extends Observable<T> implements Callable<T> { @Override public T call() throws Exception { return ObjectHelper.requireNonNull(callable.call(), "The callable returned a null value"); } ObservableFromCallable(Callable<? extends T> callable); @Override void subscribeActual(Observer<? super T> s); @Override T call();  }
@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) { 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); observerLatch.await(); outer.cancel(); funcLatch.countDown(); verify(func).call(); verify(observer).onSubscribe(any(Disposable.class)); verifyNoMoreInteractions(observer); }