@Test
public void testResumeNextWithSynchronousExecution() {

    final AtomicReference<Throwable> receivedException = new AtomicReference<Throwable>();
    Observable<String> w = Observable.create(new Observable.OnSubscribeFunc<String>() {
        @Override
        public Subscription onSubscribe(Observer<? super String> observer) {
            observer.onNext("one");
            observer.onError(new Throwable("injected failure"));
            return Subscriptions.empty();
        }
    });
    Func1<Throwable, Observable<String>> resume = new Func1<Throwable, Observable<String>>() {
        @Override
        public Observable<String> call(Throwable t1) {
            receivedException.set(t1);
            return Observable.from("twoResume", "threeResume");
        }
    };
    Observable<String> observable = Observable.create(onErrorResumeNextViaFunction(w, resume));
    @SuppressWarnings("unchecked")
    Observer<String> observer = mock(Observer.class);
    observable.subscribe(observer);
    verify(observer, Mockito.never()).onError(any(Throwable.class));
    verify(observer, times(1)).onCompleted();
    verify(observer, times(1)).onNext("one");
    verify(observer, Mockito.never()).onNext("two");
    verify(observer, Mockito.never()).onNext("three");
    verify(observer, times(1)).onNext("twoResume");
    verify(observer, times(1)).onNext("threeResume");
    assertNotNull(receivedException.get());
}