@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";
        }
    });
    Maybe<String> fromCallableObservable = Maybe.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);
}