@Test public void testDelayWithError() { Observable<Long> source = Observable.interval(1L, TimeUnit.SECONDS, scheduler).map(new Func1<Long, Long>() { @Override public Long call(Long value) { if (value == 1L) { throw new RuntimeException("error!"); } return value; } }); Observable<Long> delayed = source.delay(1L, TimeUnit.SECONDS, scheduler); delayed.subscribe(observer); InOrder inOrder = inOrder(observer); scheduler.advanceTimeTo(1999L, TimeUnit.MILLISECONDS); verify(observer, never()).onNext(anyLong()); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(2000L, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onError(any(Throwable.class)); inOrder.verify(observer, never()).onNext(anyLong()); verify(observer, never()).onCompleted(); scheduler.advanceTimeTo(5000L, TimeUnit.MILLISECONDS); inOrder.verify(observer, never()).onNext(anyLong()); inOrder.verify(observer, never()).onError(any(Throwable.class)); verify(observer, 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);  }