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);  }
@Test public void testDelay() { Observable<Long> source = Observable.interval(1L, TimeUnit.SECONDS, scheduler).take(3); Observable<Long> delayed = source.delay(500L, TimeUnit.MILLISECONDS, scheduler); delayed.subscribe(observer); InOrder inOrder = inOrder(observer); scheduler.advanceTimeTo(1499L, TimeUnit.MILLISECONDS); verify(observer, never()).onNext(anyLong()); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(1500L, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext(0L); inOrder.verify(observer, never()).onNext(anyLong()); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(2400L, TimeUnit.MILLISECONDS); inOrder.verify(observer, never()).onNext(anyLong()); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(2500L, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext(1L); inOrder.verify(observer, never()).onNext(anyLong()); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(3400L, TimeUnit.MILLISECONDS); inOrder.verify(observer, never()).onNext(anyLong()); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(3500L, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext(2L); verify(observer, times(1)).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
@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)); }
@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(); }
@Ignore @Test public void testDelayWithMultipleSubscriptions() { Observable<Long> source = Observable.interval(1L, TimeUnit.SECONDS, scheduler).take(3); Observable<Long> delayed = source.delay(500L, TimeUnit.MILLISECONDS, scheduler); delayed.subscribe(observer); delayed.subscribe(observer2); InOrder inOrder = inOrder(observer); InOrder inOrder2 = inOrder(observer2); scheduler.advanceTimeTo(1499L, TimeUnit.MILLISECONDS); verify(observer, never()).onNext(anyLong()); verify(observer2, never()).onNext(anyLong()); scheduler.advanceTimeTo(1500L, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext(0L); inOrder2.verify(observer2, times(1)).onNext(0L); scheduler.advanceTimeTo(2499L, TimeUnit.MILLISECONDS); inOrder.verify(observer, never()).onNext(anyLong()); inOrder2.verify(observer2, never()).onNext(anyLong()); scheduler.advanceTimeTo(2500L, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext(1L); inOrder2.verify(observer2, times(1)).onNext(1L); verify(observer, never()).onCompleted(); verify(observer2, never()).onCompleted(); scheduler.advanceTimeTo(3500L, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext(2L); inOrder2.verify(observer2, times(1)).onNext(2L); inOrder.verify(observer, never()).onNext(anyLong()); inOrder2.verify(observer2, never()).onNext(anyLong()); inOrder.verify(observer, times(1)).onCompleted(); inOrder2.verify(observer2, times(1)).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); verify(observer2, never()).onError(any(Throwable.class)); }
@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)); }
@Test public void testDelayWithObservableSingleSend1() { PublishSubject<Integer> source = PublishSubject.create(); final PublishSubject<Integer> delay = PublishSubject.create(); 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(delayFunc).subscribe(o); source.onNext(1); delay.onNext(1); delay.onNext(2); inOrder.verify(o).onNext(1); inOrder.verifyNoMoreInteractions(); verify(o, never()).onError(any(Throwable.class)); }
@Test public void testDelayWithObservableSourceThrows() { PublishSubject<Integer> source = PublishSubject.create(); final PublishSubject<Integer> delay = PublishSubject.create(); 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(delayFunc).subscribe(o); source.onNext(1); source.onError(new OperationReduceTest.CustomException()); delay.onNext(1); inOrder.verify(o).onError(any(OperationReduceTest.CustomException.class)); inOrder.verifyNoMoreInteractions(); verify(o, never()).onNext(any()); verify(o, never()).onCompleted(); }
@Test public void testDelayWithObservableDelayFunctionThrows() { PublishSubject<Integer> source = PublishSubject.create(); Func1<Integer, Observable<Integer>> delayFunc = new Func1<Integer, Observable<Integer>>() { @Override public Observable<Integer> call(Integer t1) { throw new OperationReduceTest.CustomException(); } }; @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); source.delay(delayFunc).subscribe(o); source.onNext(1); inOrder.verify(o).onError(any(OperationReduceTest.CustomException.class)); inOrder.verifyNoMoreInteractions(); verify(o, never()).onNext(any()); verify(o, never()).onCompleted(); }
@Test public void testDelayWithObservableDelayThrows() { PublishSubject<Integer> source = PublishSubject.create(); final PublishSubject<Integer> delay = PublishSubject.create(); 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(delayFunc).subscribe(o); source.onNext(1); delay.onError(new OperationReduceTest.CustomException()); inOrder.verify(o).onError(any(OperationReduceTest.CustomException.class)); inOrder.verifyNoMoreInteractions(); verify(o, never()).onNext(any()); verify(o, never()).onCompleted(); }
@Test public void testDelayWithObservableSubscriptionNormal() { PublishSubject<Integer> source = PublishSubject.create(); final PublishSubject<Integer> delay = PublishSubject.create(); Func0<Observable<Integer>> subFunc = new Func0<Observable<Integer>>() { @Override public Observable<Integer> call() { return delay; } }; 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); delay.onNext(1); source.onNext(2); delay.onNext(2); inOrder.verify(o).onNext(2); inOrder.verifyNoMoreInteractions(); verify(o, never()).onError(any(Throwable.class)); verify(o, never()).onCompleted(); }
@Test public void testDelayWithObservableSubscriptionFunctionThrows() { PublishSubject<Integer> source = PublishSubject.create(); final PublishSubject<Integer> delay = PublishSubject.create(); Func0<Observable<Integer>> subFunc = new Func0<Observable<Integer>>() { @Override public Observable<Integer> call() { throw new OperationReduceTest.CustomException(); } }; 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); delay.onNext(1); source.onNext(2); inOrder.verify(o).onError(any(OperationReduceTest.CustomException.class)); inOrder.verifyNoMoreInteractions(); verify(o, never()).onNext(any()); verify(o, never()).onCompleted(); }
@Test public void testDelayWithObservableSubscriptionThrows() { PublishSubject<Integer> source = PublishSubject.create(); final PublishSubject<Integer> delay = PublishSubject.create(); Func0<Observable<Integer>> subFunc = new Func0<Observable<Integer>>() { @Override public Observable<Integer> call() { return delay; } }; 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); delay.onError(new OperationReduceTest.CustomException()); source.onNext(2); inOrder.verify(o).onError(any(OperationReduceTest.CustomException.class)); inOrder.verifyNoMoreInteractions(); verify(o, never()).onNext(any()); verify(o, never()).onCompleted(); }
@Test public void testDelayWithObservableEmptyDelayer() { PublishSubject<Integer> source = PublishSubject.create(); Func1<Integer, Observable<Integer>> delayFunc = new Func1<Integer, Observable<Integer>>() { @Override public Observable<Integer> call(Integer t1) { return Observable.empty(); } }; @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); source.delay(delayFunc).subscribe(o); source.onNext(1); source.onCompleted(); inOrder.verify(o).onNext(1); inOrder.verify(o).onCompleted(); inOrder.verifyNoMoreInteractions(); verify(o, never()).onError(any(Throwable.class)); }
@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(); }