MaybeFromCallable extends Maybe<T> implements Callable<T> { @Override public T call() throws Exception { return callable.call(); } MaybeFromCallable(Callable<? extends T> callable); @Override T call();  }
@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"; } }); Maybe<String> fromCallableObservable = Maybe.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); }