@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()); }
public static <T> OnSubscribeFunc<T> onErrorResumeNextViaFunction(Observable<? extends T> originalSequence, Func1<Throwable, ? extends Observable<? extends T>> resumeFunction) { return new OnErrorResumeNextViaFunction<T>(originalSequence, resumeFunction); }
OperationOnErrorResumeNextViaFunction { public static <T> OnSubscribeFunc<T> onErrorResumeNextViaFunction(Observable<? extends T> originalSequence, Func1<Throwable, ? extends Observable<? extends T>> resumeFunction) { return new OnErrorResumeNextViaFunction<T>(originalSequence, resumeFunction); } }
OperationOnErrorResumeNextViaFunction { public static <T> OnSubscribeFunc<T> onErrorResumeNextViaFunction(Observable<? extends T> originalSequence, Func1<Throwable, ? extends Observable<? extends T>> resumeFunction) { return new OnErrorResumeNextViaFunction<T>(originalSequence, resumeFunction); }  }
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); }
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);  }