@Test
public void testResumeNextWithAsyncExecution() {

    final AtomicReference<Throwable> receivedException = new AtomicReference<Throwable>();
    Subscription s = mock(Subscription.class);
    TestObservable w = new TestObservable(s, "one");
    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(Observable.create(w), resume));
    @SuppressWarnings("unchecked")
    Observer<String> observer = mock(Observer.class);
    observable.subscribe(observer);
    try {
        w.t.join();
    } catch (InterruptedException e) {
        fail(e.getMessage());
    }
    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());
}