OperationFromFunctionals { public static <R> OnSubscribeFunc<R> fromRunnable(final Runnable run, final R result) { return new InvokeAsync(new Func0<R>() { @Override public R call() { run.run(); return result; } }); } private  OperationFromFunctionals(); static OnSubscribeFunc<R> fromAction(Action0 action, R result); static OnSubscribeFunc<R> fromFunc0(Func0<? extends R> function); static OnSubscribeFunc<R> fromCallable(Callable<? extends R> callable); static OnSubscribeFunc<R> fromRunnable(final Runnable run, final R result);  }
@Test public void testFromRunnable() { final AtomicInteger value = new AtomicInteger(); Runnable action = new Runnable() { @Override public void run() { value.set(2); } }; Observable<Integer> source = Async.fromRunnable(action, 1, scheduler); for (int i = 0; i < 3; i++) { value.set(0); Observer<Object> observer = mock(Observer.class); source.subscribe(new TestObserver<Object>(observer)); InOrder inOrder = inOrder(observer); inOrder.verify(observer, never()).onNext(any()); inOrder.verify(observer, never()).onCompleted(); scheduler.advanceTimeBy(1, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext(1); inOrder.verify(observer, times(1)).onCompleted(); inOrder.verifyNoMoreInteractions(); verify(observer, never()).onError(any(Throwable.class)); Assert.assertEquals(2, value.get()); } }
@Test public void testFromRunnableThrows() { Runnable action = new Runnable() { @Override public void run() { throw new RuntimeException("Forced failure!"); } }; Observable<Integer> source = Async.fromRunnable(action, 1, scheduler); testRunShouldThrow(source, RuntimeException.class); }