@Test
public void testFailingComputations() throws Exception {

    AtomicBoolean flag = new AtomicBoolean();
    ComputationTaskInvoker<String> slowInvoker = new ComputationTaskInvoker<>(
            Observable.create(subscriber -> {
                if (flag.getAndSet(true)) {
                    subscriber.onNext("OK!");
                    subscriber.onCompleted();
                } else {
                    Observable.<String>timer(1, TimeUnit.SECONDS, testScheduler)
                            .flatMap(tick -> Observable.<String>error(new RuntimeException("simulated computation error")))
                            .subscribe(subscriber);
                }
            }),
            testScheduler
    );
    ExtTestSubscriber<String> failingSubscriber = new ExtTestSubscriber<>();
    slowInvoker.recompute().subscribe(failingSubscriber);
    testScheduler.triggerActions();
    ExtTestSubscriber<String> okSubscriber = new ExtTestSubscriber<>();
    slowInvoker.recompute().subscribe(okSubscriber);
    testScheduler.advanceTimeBy(1, TimeUnit.SECONDS);
    failingSubscriber.assertOnError();
    assertThat(okSubscriber.takeNext()).isEqualTo("OK!");
    okSubscriber.assertOnCompleted();
}