OperationOnErrorReturn { public static <T> OnSubscribeFunc<T> onErrorReturn(Observable<? extends T> originalSequence, Func1<Throwable, ? extends T> resumeFunction) { return new OnErrorReturn<T>(originalSequence, resumeFunction); }  static OnSubscribeFunc<T> onErrorReturn(Observable<? extends T> originalSequence, Func1<Throwable, ? extends T> resumeFunction);  }
@Test public void testResumeNext() { 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); return "failure"; } })); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); observable.subscribe(observer); try { f.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, times(1)).onNext("failure"); assertNotNull(capturedException.get()); }
@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()); } verify(observer, times(1)).onNext("one"); verify(observer, times(1)).onError(any(Throwable.class)); verify(observer, times(0)).onCompleted(); assertNotNull(capturedException.get()); }