@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)); }
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); }
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); } }
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(); }
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); }
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);  }