OperationDelay { public static <T> OnSubscribeFunc<T> delaySubscription(Observable<? extends T> source, long time, TimeUnit unit, Scheduler scheduler) { return new DelaySubscribeFunc<T>(source, time, unit, scheduler); }  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 testDelaySubscription() { TestScheduler scheduler = new TestScheduler(); Observable<Integer> result = Observable.from(1, 2, 3).delaySubscription(100, TimeUnit.MILLISECONDS, scheduler); Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); result.subscribe(o); inOrder.verify(o, never()).onNext(any()); inOrder.verify(o, never()).onCompleted(); scheduler.advanceTimeBy(100, TimeUnit.MILLISECONDS); inOrder.verify(o, times(1)).onNext(1); inOrder.verify(o, times(1)).onNext(2); inOrder.verify(o, times(1)).onNext(3); inOrder.verify(o, times(1)).onCompleted(); verify(o, never()).onError(any(Throwable.class)); }
@Test public void testDelaySubscriptionCancelBeforeTime() { TestScheduler scheduler = new TestScheduler(); Observable<Integer> result = Observable.from(1, 2, 3).delaySubscription(100, TimeUnit.MILLISECONDS, scheduler); Observer<Object> o = mock(Observer.class); Subscription s = result.subscribe(o); s.unsubscribe(); scheduler.advanceTimeBy(100, TimeUnit.MILLISECONDS); verify(o, never()).onNext(any()); verify(o, never()).onCompleted(); verify(o, never()).onError(any(Throwable.class)); }