@Test public void testLongDelay() { Observable<Long> source = Observable.interval(1L, TimeUnit.SECONDS, scheduler).take(3); Observable<Long> delayed = source.delay(5L, TimeUnit.SECONDS, scheduler); delayed.subscribe(observer); InOrder inOrder = inOrder(observer); scheduler.advanceTimeTo(5999L, TimeUnit.MILLISECONDS); verify(observer, never()).onNext(anyLong()); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(6000L, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext(0L); scheduler.advanceTimeTo(6999L, TimeUnit.MILLISECONDS); inOrder.verify(observer, never()).onNext(anyLong()); scheduler.advanceTimeTo(7000L, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext(1L); scheduler.advanceTimeTo(7999L, TimeUnit.MILLISECONDS); inOrder.verify(observer, never()).onNext(anyLong()); scheduler.advanceTimeTo(8000L, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext(2L); inOrder.verify(observer, times(1)).onCompleted(); inOrder.verify(observer, never()).onNext(anyLong()); inOrder.verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
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);  }