OperationStartFuture { public static <T> Observable<T> startFuture(Func0<? extends Future<? extends T>> functionAsync) { Future<? extends T> task; try { task = functionAsync.call(); } catch (Throwable t) { return Observable.error(t); } return Observable.from(task); } private  OperationStartFuture(); static Observable<T> startFuture(Func0<? extends Future<? extends T>> functionAsync); static Observable<T> startFuture(Func0<? extends Future<? extends T>> functionAsync,
            Scheduler scheduler);  }
@Test @SuppressWarnings("unchecked") public void testSimple() throws InterruptedException { final ExecutorService exec = Executors.newCachedThreadPool(); try { final CountDownLatch ready = new CountDownLatch(1); Func0<Future<Integer>> func = new Func0<Future<Integer>>() { @Override public Future<Integer> call() { return exec.submit(new Callable<Integer>() { @Override public Integer call() throws Exception { if (!ready.await(1000, TimeUnit.MILLISECONDS)) { throw new IllegalStateException("Not started in time"); } return 1; } }); } }; Observable<Integer> result = Async.startFuture(func, Schedulers.threadPoolForComputation()); final Observer<Integer> observer = mock(Observer.class); final CountDownLatch done = new CountDownLatch(1); result.subscribe(new 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(); } }
@Test @SuppressWarnings("unchecked") public void testSimpleFactoryThrows() { Func0<Future<Integer>> func = new Func0<Future<Integer>>() { @Override public Future<Integer> call() { throw new CustomException(); } }; Observable<Integer> result = Async.startFuture(func); final Observer<Object> observer = mock(Observer.class); result.subscribe(observer); verify(observer, never()).onNext(any()); verify(observer, never()).onCompleted(); verify(observer).onError(any(CustomException.class)); }