@Test @SuppressWarnings("unchecked") public void testSimple() throws InterruptedException { final ExecutorService exec = Executors.newCachedThreadPool(); try { final CountDownLatch ready = new CountDownLatch(1); Func0<Future<Observable<Integer>>> func = new Func0<Future<Observable<Integer>>>() { @Override public Future<Observable<Integer>> call() { return exec.submit(new Callable<Observable<Integer>>() { @Override public Observable<Integer> call() throws Exception { if (!ready.await(1000, TimeUnit.MILLISECONDS)) { throw new IllegalStateException("Not started in time"); } return Observable.from(1); } }); } }; Observable<Integer> result = Async.deferFuture(func, Schedulers.threadPoolForComputation()); final Observer<Integer> observer = mock(Observer.class); final CountDownLatch done = new CountDownLatch(1); result.subscribe(new OperationStartFutureTest.MockHelper<Integer>(observer, done)); ready.countDown(); if (!done.await(1000, TimeUnit.MILLISECONDS)) { fail("Not completed in time!"); } InOrder inOrder = inOrder(observer); inOrder.verify(observer).onNext(1); inOrder.verify(observer).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); } finally { exec.shutdown(); } }
public static <T> Observable<T> deferFuture(Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync) { return Observable.defer(new DeferFutureFunc0<T>(observableFactoryAsync)); }
OperationDeferFuture { public static <T> Observable<T> deferFuture(Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync) { return Observable.defer(new DeferFutureFunc0<T>(observableFactoryAsync)); } }
OperationDeferFuture { public static <T> Observable<T> deferFuture(Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync) { return Observable.defer(new DeferFutureFunc0<T>(observableFactoryAsync)); } private  OperationDeferFuture(); }
OperationDeferFuture { public static <T> Observable<T> deferFuture(Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync) { return Observable.defer(new DeferFutureFunc0<T>(observableFactoryAsync)); } private  OperationDeferFuture(); 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); }
OperationDeferFuture { public static <T> Observable<T> deferFuture(Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync) { return Observable.defer(new DeferFutureFunc0<T>(observableFactoryAsync)); } private  OperationDeferFuture(); 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);  }