@Test
public void testSimpleScheduledThrowing() {

    
    Observable<Integer> source = Observable.<Integer>error(new CustomException())
            .subscribeOn(Schedulers.threadPoolForComputation());
    
    final AtomicInteger sum = new AtomicInteger();
    Action1<Integer> add = new Action1<Integer>() {
        @Override
        public void call(Integer t1) {
            sum.addAndGet(t1);
        }
    };
    
    FutureTask<Void> task = Async.forEachFuture(source, add, Schedulers.newThread());
    
    try {
        task.get(1000, TimeUnit.MILLISECONDS);
    } catch (TimeoutException ex) {
        fail("Timed out: " + ex);
    } catch (ExecutionException ex) {
        if (!(ex.getCause() instanceof CustomException)) {
            fail("Got different exception: " + ex.getCause());
        }
    } catch (InterruptedException ex) {
        fail("Exception: " + ex);
    }
    
    assertEquals(0, sum.get());
}