@Test
public void testFunctionThrowsError() {

    Subscription s = mock(Subscription.class);
    TestObservable f = new TestObservable(s, "one");
    Observable<String> w = Observable.create(f);
    final AtomicReference<Throwable> capturedException = new AtomicReference<Throwable>();
    Observable<String> observable = Observable.create(onErrorReturn(w, new Func1<Throwable, String>() {
        @Override
        public String call(Throwable e) {
            capturedException.set(e);
            throw new RuntimeException("exception from function");
        }
    }));
    @SuppressWarnings("unchecked")
    Observer<String> observer = mock(Observer.class);
    observable.subscribe(observer);
    try {
        f.t.join();
    } catch (InterruptedException e) {
        fail(e.getMessage());
    }
    // we should get the "one" value before the error
    verify(observer, times(1)).onNext("one");
    // we should have received an onError call on the Observer since the resume function threw an exception
    verify(observer, times(1)).onError(any(Throwable.class));
    verify(observer, times(0)).onCompleted();
    assertNotNull(capturedException.get());
}