OperationOnErrorResumeNextViaFunction { public static <T> OnSubscribeFunc<T> onErrorResumeNextViaFunction(Observable<? extends T> originalSequence, Func1<Throwable, ? extends Observable<? extends T>> resumeFunction) { return new OnErrorResumeNextViaFunction<T>(originalSequence, resumeFunction); }  static OnSubscribeFunc<T> onErrorResumeNextViaFunction(Observable<? extends T> originalSequence, Func1<Throwable, ? extends Observable<? extends T>> resumeFunction);  }
@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()); }
@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()); }
@Test public void testFunctionThrowsError() { 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) { throw new RuntimeException("exception from function"); } }; 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, times(1)).onNext("one"); verify(observer, times(1)).onError(any(Throwable.class)); verify(observer, times(0)).onCompleted(); }