@Test
public void testDelayWithObservableNormal1() {

    PublishSubject<Integer> source = PublishSubject.create();
    final List<PublishSubject<Integer>> delays = new ArrayList<PublishSubject<Integer>>();
    final int n = 10;
    for (int i = 0; i < n; i++) {
        PublishSubject<Integer> delay = PublishSubject.create();
        delays.add(delay);
    }
    Func1<Integer, Observable<Integer>> delayFunc = new Func1<Integer, Observable<Integer>>() {
        @Override
        public Observable<Integer> call(Integer t1) {
            return delays.get(t1);
        }
    };
    @SuppressWarnings("unchecked")
    Observer<Object> o = mock(Observer.class);
    InOrder inOrder = inOrder(o);
    source.delay(delayFunc).subscribe(o);
    for (int i = 0; i < n; i++) {
        source.onNext(i);
        delays.get(i).onNext(i);
        inOrder.verify(o).onNext(i);
    }
    source.onCompleted();
    inOrder.verify(o).onCompleted();
    inOrder.verifyNoMoreInteractions();
    verify(o, never()).onError(any(Throwable.class));
}