OperationOnExceptionResumeNextViaObservable { public static <T> OnSubscribeFunc<T> onExceptionResumeNextViaObservable(Observable<? extends T> originalSequence, Observable<? extends T> resumeSequence) { return new OnExceptionResumeNextViaObservable<T>(originalSequence, resumeSequence); }  static OnSubscribeFunc<T> onExceptionResumeNextViaObservable(Observable<? extends T> originalSequence, Observable<? extends T> resumeSequence);  }
@Test public void testResumeNextWithException() { Subscription s = mock(Subscription.class); TestObservable f = new TestObservable(s, "one", "EXCEPTION", "two", "three"); Observable<String> w = Observable.create(f); Observable<String> resume = Observable.from("twoResume", "threeResume"); Observable<String> observable = Observable.create(onExceptionResumeNextViaObservable(w, resume)); @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, Mockito.never()).onNext("two"); verify(observer, Mockito.never()).onNext("three"); verify(observer, times(1)).onNext("twoResume"); verify(observer, times(1)).onNext("threeResume"); verify(observer, Mockito.never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); verifyNoMoreInteractions(observer); }
@Test public void testResumeNextWithRuntimeException() { Subscription s = mock(Subscription.class); TestObservable f = new TestObservable(s, "one", "RUNTIMEEXCEPTION", "two", "three"); Observable<String> w = Observable.create(f); Observable<String> resume = Observable.from("twoResume", "threeResume"); Observable<String> observable = Observable.create(onExceptionResumeNextViaObservable(w, resume)); @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, Mockito.never()).onNext("two"); verify(observer, Mockito.never()).onNext("three"); verify(observer, times(1)).onNext("twoResume"); verify(observer, times(1)).onNext("threeResume"); verify(observer, Mockito.never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); verifyNoMoreInteractions(observer); }
@Test public void testThrowablePassesThru() { Subscription s = mock(Subscription.class); TestObservable f = new TestObservable(s, "one", "THROWABLE", "two", "three"); Observable<String> w = Observable.create(f); Observable<String> resume = Observable.from("twoResume", "threeResume"); Observable<String> observable = Observable.create(onExceptionResumeNextViaObservable(w, resume)); @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, never()).onNext("two"); verify(observer, never()).onNext("three"); verify(observer, never()).onNext("twoResume"); verify(observer, never()).onNext("threeResume"); verify(observer, times(1)).onError(any(Throwable.class)); verify(observer, never()).onCompleted(); verifyNoMoreInteractions(observer); }
@Test public void testErrorPassesThru() { Subscription s = mock(Subscription.class); TestObservable f = new TestObservable(s, "one", "ERROR", "two", "three"); Observable<String> w = Observable.create(f); Observable<String> resume = Observable.from("twoResume", "threeResume"); Observable<String> observable = Observable.create(onExceptionResumeNextViaObservable(w, resume)); @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, never()).onNext("two"); verify(observer, never()).onNext("three"); verify(observer, never()).onNext("twoResume"); verify(observer, never()).onNext("threeResume"); verify(observer, times(1)).onError(any(Throwable.class)); verify(observer, never()).onCompleted(); verifyNoMoreInteractions(observer); }
@Test public void testMapResumeAsyncNext() { Subscription sr = mock(Subscription.class); Observable<String> w = Observable.from("one", "fail", "two", "three", "fail"); TestObservable f = new TestObservable(sr, "twoResume", "threeResume"); Observable<String> resume = Observable.create(f); w = w.map(new Func1<String, String>() { public String call(String s) { if ("fail".equals(s)) throw new RuntimeException("Forced Failure"); System.out.println("BadMapper:" + s); return s; } }); Observable<String> observable = Observable.create(onExceptionResumeNextViaObservable(w, resume)); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); observable.subscribe(observer); try { if (f.t != null) { f.t.join(); } } catch (InterruptedException e) { fail(e.getMessage()); } verify(observer, times(1)).onNext("one"); verify(observer, never()).onNext("two"); verify(observer, never()).onNext("three"); verify(observer, times(1)).onNext("twoResume"); verify(observer, times(1)).onNext("threeResume"); verify(observer, Mockito.never()).onError(any(Throwable.class)); verify(observer, times(1)).onCompleted(); }