OperationForEachFuture { public static <T> FutureTask<Void> forEachFuture( Observable<? extends T> source, Action1<? super T> onNext) { return forEachFuture(source, onNext, Functionals.emptyThrowable(), Functionals.empty()); } private  OperationForEachFuture(); static FutureTask<Void> forEachFuture(
            Observable<? extends T> source, 
            Action1<? super T> onNext); static FutureTask<Void> forEachFuture(
            Observable<? extends T> source, 
            Action1<? super T> onNext,
            Action1<? super Throwable> onError); static FutureTask<Void> forEachFuture(
            Observable<? extends T> source, 
            Action1<? super T> onNext,
            Action1<? super Throwable> onError,
            Action0 onCompleted);  }
@Test public void testSimple() { final ExecutorService exec = Executors.newCachedThreadPool(); try { Observable<Integer> source = Observable.from(1, 2, 3) .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); exec.execute(task); try { Void value = task.get(1000, TimeUnit.MILLISECONDS); assertEquals(null, value); assertEquals(6, sum.get()); } catch (TimeoutException ex) { fail("Timed out: " + ex); } catch (ExecutionException ex) { fail("Exception: " + ex); } catch (InterruptedException ex) { fail("Exception: " + ex); } } finally { exec.shutdown(); } }
@Test public void testSimpleThrowing() { final ExecutorService exec = Executors.newCachedThreadPool(); try { 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); exec.execute(task); 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()); } finally { exec.shutdown(); } }
@Test public void testSimpleScheduled() { Observable<Integer> source = Observable.from(1, 2, 3) .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 { Void value = task.get(1000, TimeUnit.MILLISECONDS); assertEquals(null, value); assertEquals(6, sum.get()); } catch (TimeoutException ex) { fail("Timed out: " + ex); } catch (ExecutionException ex) { fail("Exception: " + ex); } catch (InterruptedException ex) { fail("Exception: " + ex); } }
@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()); }