@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(); }
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);  }