@Test public void testPhasedRetry() { final int count = 100; AbstractOnSubscribe<String, Void> aos = new AbstractOnSubscribe<String, Void>() { int calls; int phase; @Override protected void next(SubscriptionState<String, Void> state) { switch (phase) { case 0: if (calls++ < count) { state.onNext("Beginning"); state.onError(new TestException()); } else { phase++; } break; case 1: state.onNext("Beginning"); phase++; break; case 2: state.onNext("Finally"); state.onCompleted(); phase++; break; default: throw new IllegalStateException("Wrong phase: " + state.phase()); } } }; @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); aos.toObservable().retry(2 * count).subscribe(o); verify(o, never()).onError(any(Throwable.class)); inOrder.verify(o, times(count + 1)).onNext("Beginning"); inOrder.verify(o).onNext("Finally"); inOrder.verify(o).onCompleted(); inOrder.verifyNoMoreInteractions(); }
public final Observable<T> toObservable() { return Observable.create(this); }
AbstractOnSubscribe implements OnSubscribe<T> { public final Observable<T> toObservable() { return Observable.create(this); } }
AbstractOnSubscribe implements OnSubscribe<T> { public final Observable<T> toObservable() { return Observable.create(this); }  }
AbstractOnSubscribe implements OnSubscribe<T> { public final Observable<T> toObservable() { return Observable.create(this); }  @Override final void call(final Subscriber<? super T> subscriber); final Observable<T> toObservable(); static AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next); static AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next,
            Func1<? super Subscriber<? super T>, ? extends S> onSubscribe); static AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next,
            Func1<? super Subscriber<? super T>, ? extends S> onSubscribe, Action1<? super S> onTerminated); }
AbstractOnSubscribe implements OnSubscribe<T> { public final Observable<T> toObservable() { return Observable.create(this); }  @Override final void call(final Subscriber<? super T> subscriber); final Observable<T> toObservable(); static AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next); static AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next,
            Func1<? super Subscriber<? super T>, ? extends S> onSubscribe); static AbstractOnSubscribe<T, S> create(Action1<SubscriptionState<T, S>> next,
            Func1<? super Subscriber<? super T>, ? extends S> onSubscribe, Action1<? super S> onTerminated);  }