@Test public void testDelayWithObservableSubscriptionRunCompletion() { PublishSubject<Integer> source = PublishSubject.create(); final PublishSubject<Integer> sdelay = PublishSubject.create(); final PublishSubject<Integer> delay = PublishSubject.create(); Func0<Observable<Integer>> subFunc = new Func0<Observable<Integer>>() { @Override public Observable<Integer> call() { return sdelay; } }; Func1<Integer, Observable<Integer>> delayFunc = new Func1<Integer, Observable<Integer>>() { @Override public Observable<Integer> call(Integer t1) { return delay; } }; @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); source.delay(subFunc, delayFunc).subscribe(o); source.onNext(1); sdelay.onCompleted(); source.onNext(2); delay.onNext(2); inOrder.verify(o).onNext(2); inOrder.verifyNoMoreInteractions(); verify(o, never()).onError(any(Throwable.class)); verify(o, never()).onCompleted(); }
public static <T> Observable<T> delay(Observable<T> observable, final long delay, final TimeUnit unit, final Scheduler scheduler) { Observable<Observable<T>> seqs = observable.map(new Func1<T, Observable<T>>() { public Observable<T> call(final T x) { ConnectableObservable<T> co = Observable.timer(delay, unit, scheduler).map(new Func1<Long, T>() { @Override public T call(Long ignored) { return x; } }).replay(); co.connect(); return co; } }); return Observable.concat(seqs); }
OperationDelay { public static <T> Observable<T> delay(Observable<T> observable, final long delay, final TimeUnit unit, final Scheduler scheduler) { Observable<Observable<T>> seqs = observable.map(new Func1<T, Observable<T>>() { public Observable<T> call(final T x) { ConnectableObservable<T> co = Observable.timer(delay, unit, scheduler).map(new Func1<Long, T>() { @Override public T call(Long ignored) { return x; } }).replay(); co.connect(); return co; } }); return Observable.concat(seqs); } }
OperationDelay { public static <T> Observable<T> delay(Observable<T> observable, final long delay, final TimeUnit unit, final Scheduler scheduler) { Observable<Observable<T>> seqs = observable.map(new Func1<T, Observable<T>>() { public Observable<T> call(final T x) { ConnectableObservable<T> co = Observable.timer(delay, unit, scheduler).map(new Func1<Long, T>() { @Override public T call(Long ignored) { return x; } }).replay(); co.connect(); return co; } }); return Observable.concat(seqs); }  }
OperationDelay { public static <T> Observable<T> delay(Observable<T> observable, final long delay, final TimeUnit unit, final Scheduler scheduler) { Observable<Observable<T>> seqs = observable.map(new Func1<T, Observable<T>>() { public Observable<T> call(final T x) { ConnectableObservable<T> co = Observable.timer(delay, unit, scheduler).map(new Func1<Long, T>() { @Override public T call(Long ignored) { return x; } }).replay(); co.connect(); return co; } }); return Observable.concat(seqs); }  static Observable<T> delay(Observable<T> observable, final long delay, final TimeUnit unit, final Scheduler scheduler); static OnSubscribeFunc<T> delaySubscription(Observable<? extends T> source, long time, TimeUnit unit, Scheduler scheduler); static OnSubscribeFunc<T> delay(Observable<? extends T> source,
            Func1<? super T, ? extends Observable<U>> itemDelay); static OnSubscribeFunc<T> delay(Observable<? extends T> source,
            Func0<? extends Observable<U>> subscriptionDelay,
            Func1<? super T, ? extends Observable<V>> itemDelay); }
OperationDelay { public static <T> Observable<T> delay(Observable<T> observable, final long delay, final TimeUnit unit, final Scheduler scheduler) { Observable<Observable<T>> seqs = observable.map(new Func1<T, Observable<T>>() { public Observable<T> call(final T x) { ConnectableObservable<T> co = Observable.timer(delay, unit, scheduler).map(new Func1<Long, T>() { @Override public T call(Long ignored) { return x; } }).replay(); co.connect(); return co; } }); return Observable.concat(seqs); }  static Observable<T> delay(Observable<T> observable, final long delay, final TimeUnit unit, final Scheduler scheduler); static OnSubscribeFunc<T> delaySubscription(Observable<? extends T> source, long time, TimeUnit unit, Scheduler scheduler); static OnSubscribeFunc<T> delay(Observable<? extends T> source,
            Func1<? super T, ? extends Observable<U>> itemDelay); static OnSubscribeFunc<T> delay(Observable<? extends T> source,
            Func0<? extends Observable<U>> subscriptionDelay,
            Func1<? super T, ? extends Observable<V>> itemDelay);  }