@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); } }
public static <T> FutureTask<Void> forEachFuture( Observable<? extends T> source, Action1<? super T> onNext) { return forEachFuture(source, onNext, Functionals.emptyThrowable(), Functionals.empty()); }
OperationForEachFuture { public static <T> FutureTask<Void> forEachFuture( Observable<? extends T> source, Action1<? super T> onNext) { return forEachFuture(source, onNext, Functionals.emptyThrowable(), Functionals.empty()); } }
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(); }
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); }
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);  }