Async { public static <T> StoppableObservable<T> runAsync(Scheduler scheduler, final Action2<? super Observer<? super T>, ? super Subscription> action) { return runAsync(scheduler, PublishSubject.<T>create(), action); } private  Async(); static Observable<T> start(Func0<T> func); static Observable<T> start(Func0<T> func, Scheduler scheduler); static Func0<Observable<Void>> toAsync(Action0 action); static Func0<Observable<R>> toAsync(Func0<? extends R> func); static Func1<T1, Observable<Void>> toAsync(Action1<? super T1> action); static Func1<T1, Observable<R>> toAsync(Func1<? super T1, ? extends R> func); static Func2<T1, T2, Observable<Void>> toAsync(Action2<? super T1, ? super T2> action); static Func2<T1, T2, Observable<R>> toAsync(Func2<? super T1, ? super T2, ? extends R> func); static Func3<T1, T2, T3, Observable<Void>> toAsync(Action3<? super T1, ? super T2, ? super T3> action); static Func3<T1, T2, T3, Observable<R>> toAsync(Func3<? super T1, ? super T2, ? super T3, ? extends R> func); static Func4<T1, T2, T3, T4, Observable<Void>> toAsync(Action4<? super T1, ? super T2, ? super T3, ? super T4> action); static Func4<T1, T2, T3, T4, Observable<R>> toAsync(Func4<? super T1, ? super T2, ? super T3, ? super T4, ? extends R> func); static Func5<T1, T2, T3, T4, T5, Observable<Void>> toAsync(Action5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5> action); static Func5<T1, T2, T3, T4, T5, Observable<R>> toAsync(Func5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? extends R> func); static Func6<T1, T2, T3, T4, T5, T6, Observable<Void>> toAsync(Action6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6> action); static Func6<T1, T2, T3, T4, T5, T6, Observable<R>> toAsync(Func6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? extends R> func); static Func7<T1, T2, T3, T4, T5, T6, T7, Observable<Void>> toAsync(Action7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7> action); static Func7<T1, T2, T3, T4, T5, T6, T7, Observable<R>> toAsync(Func7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? extends R> func); static Func8<T1, T2, T3, T4, T5, T6, T7, T8, Observable<Void>> toAsync(Action8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8> action); static Func8<T1, T2, T3, T4, T5, T6, T7, T8, Observable<R>> toAsync(Func8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? extends R> func); static Func9<T1, T2, T3, T4, T5, T6, T7, T8, T9, Observable<Void>> toAsync(Action9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9> action); static Func9<T1, T2, T3, T4, T5, T6, T7, T8, T9, Observable<R>> toAsync(Func9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9, ? extends R> func); static FuncN<Observable<Void>> toAsync(ActionN action); static FuncN<Observable<R>> toAsync(FuncN<? extends R> func); static Func0<Observable<Void>> toAsync(final Action0 action, final Scheduler scheduler); static Func0<Observable<R>> toAsync(final Func0<? extends R> func, final Scheduler scheduler); static Func1<T1, Observable<Void>> toAsync(final Action1<? super T1> action, final Scheduler scheduler); static Func1<T1, Observable<R>> toAsync(final Func1<? super T1, ? extends R> func, final Scheduler scheduler); static Func2<T1, T2, Observable<Void>> toAsync(final Action2<? super T1, ? super T2> action, final Scheduler scheduler); static Func2<T1, T2, Observable<R>> toAsync(final Func2<? super T1, ? super T2, ? extends R> func, final Scheduler scheduler); static Func3<T1, T2, T3, Observable<Void>> toAsync(final Action3<? super T1, ? super T2, ? super T3> action, final Scheduler scheduler); static Func3<T1, T2, T3, Observable<R>> toAsync(final Func3<? super T1, ? super T2, ? super T3, ? extends R> func, final Scheduler scheduler); static Func4<T1, T2, T3, T4, Observable<Void>> toAsync(final Action4<? super T1, ? super T2, ? super T3, ? super T4> action, final Scheduler scheduler); static Func4<T1, T2, T3, T4, Observable<R>> toAsync(final Func4<? super T1, ? super T2, ? super T3, ? super T4, ? extends R> func, final Scheduler scheduler); static Func5<T1, T2, T3, T4, T5, Observable<Void>> toAsync(final Action5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5> action, final Scheduler scheduler); static Func5<T1, T2, T3, T4, T5, Observable<R>> toAsync(final Func5<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? extends R> func, final Scheduler scheduler); static Func6<T1, T2, T3, T4, T5, T6, Observable<Void>> toAsync(final Action6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6> action, final Scheduler scheduler); static Func6<T1, T2, T3, T4, T5, T6, Observable<R>> toAsync(final Func6<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? extends R> func, final Scheduler scheduler); static Func7<T1, T2, T3, T4, T5, T6, T7, Observable<Void>> toAsync(final Action7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7> action, final Scheduler scheduler); static Func7<T1, T2, T3, T4, T5, T6, T7, Observable<R>> toAsync(final Func7<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? extends R> func, final Scheduler scheduler); static Func8<T1, T2, T3, T4, T5, T6, T7, T8, Observable<Void>> toAsync(final Action8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8> action, final Scheduler scheduler); static Func8<T1, T2, T3, T4, T5, T6, T7, T8, Observable<R>> toAsync(final Func8<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? extends R> func, final Scheduler scheduler); static Func9<T1, T2, T3, T4, T5, T6, T7, T8, T9, Observable<Void>> toAsync(final Action9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9> action, final Scheduler scheduler); static Func9<T1, T2, T3, T4, T5, T6, T7, T8, T9, Observable<R>> toAsync(final Func9<? super T1, ? super T2, ? super T3, ? super T4, ? super T5, ? super T6, ? super T7, ? super T8, ? super T9, ? extends R> func, final Scheduler scheduler); static FuncN<Observable<Void>> toAsync(final ActionN action, final Scheduler scheduler); static FuncN<Observable<R>> toAsync(final FuncN<? extends R> func, final Scheduler scheduler); static FuncN<Observable<Void>> asyncAction(final ActionN action); static FuncN<Observable<Void>> asyncAction(final ActionN action, final Scheduler scheduler); static FuncN<Observable<R>> asyncFunc(final FuncN<? extends R> func); static FuncN<Observable<R>> asyncFunc(final FuncN<? extends R> func, final Scheduler scheduler); static Observable<T> startFuture(Func0<? extends Future<? extends T>> functionAsync); static Observable<T> startFuture(Func0<? extends Future<? extends T>> functionAsync,
        Scheduler scheduler); static Observable<T> deferFuture(Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync); static Observable<T> deferFuture(
        Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync,
        Scheduler scheduler); 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); static FutureTask<Void> forEachFuture(
        Observable<? extends T> source,
        Action1<? super T> onNext,
        Scheduler scheduler); static FutureTask<Void> forEachFuture(
        Observable<? extends T> source,
        Action1<? super T> onNext,
        Action1<? super Throwable> onError,
        Scheduler scheduler); static FutureTask<Void> forEachFuture(
            Observable<? extends T> source,
            Action1<? super T> onNext,
            Action1<? super Throwable> onError,
            Action0 onCompleted,
            Scheduler scheduler); static Observable<R> fromAction(Action0 action, R result); static Observable<R> fromFunc0(Func0<? extends R> function); static Observable<R> fromCallable(Callable<? extends R> callable); static Observable<R> fromRunnable(final Runnable run, final R result); static Observable<R> fromAction(Action0 action, R result, Scheduler scheduler); static Observable<R> fromFunc0(Func0<? extends R> function, Scheduler scheduler); static Observable<R> fromCallable(Callable<? extends R> callable, Scheduler scheduler); static Observable<R> fromRunnable(final Runnable run, final R result, Scheduler scheduler); static StoppableObservable<T> runAsync(Scheduler scheduler, 
            final Action2<? super Observer<? super T>, ? super Subscription> action); static StoppableObservable<U> runAsync(Scheduler scheduler,
            final Subject<T, U> subject, 
            final Action2<? super Observer<? super T>, ? super Subscription> action);  }
@Test public void testRunAsync() throws InterruptedException { final CountDownLatch cdl = new CountDownLatch(1); final CountDownLatch cdl2 = new CountDownLatch(1); Action2<Observer<? super Integer>, Subscription> action = new Action2<Observer<? super Integer>, Subscription>() { @Override public void call(Observer<? super Integer> t1, Subscription t2) { try { cdl.await(); } catch (InterruptedException ex) { Thread.currentThread().interrupt(); return; } for (int i = 0; i < 10 && !t2.isUnsubscribed(); i++) { t1.onNext(i); } t1.onCompleted(); cdl2.countDown(); } }; @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); StoppableObservable<Integer> so = Async.<Integer>runAsync(Schedulers.io(), action); so.subscribe(o); cdl.countDown(); if (!cdl2.await(2, TimeUnit.SECONDS)) { fail("Didn't complete"); } for (int i = 0; i < 10; i++) { inOrder.verify(o).onNext(i); } inOrder.verify(o).onCompleted(); inOrder.verifyNoMoreInteractions(); verify(o, never()).onError(any(Throwable.class)); }