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