Merge pull request ReactiveX#576 from samuelgruetter/delay2

benjchristensen · benjchristensen · commit 20f50d3eff64 · 2013-12-10T14:59:02.000-08:00
Timer and Delay
diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
@@ -45,6 +45,7 @@
 import rx.operators.OperationDebounce;
 import rx.operators.OperationDefaultIfEmpty;
 import rx.operators.OperationDefer;
+import rx.operators.OperationDelay;
 import rx.operators.OperationDematerialize;
 import rx.operators.OperationDistinct;
 import rx.operators.OperationDistinctUntilChanged;
@@ -92,6 +93,7 @@
 import rx.operators.OperationThrottleFirst;
 import rx.operators.OperationTimeInterval;
 import rx.operators.OperationTimeout;
+import rx.operators.OperationTimer;
 import rx.operators.OperationTimestamp;
 import rx.operators.OperationToMap;
 import rx.operators.OperationToMultimap;
@@ -1989,6 +1991,62 @@ public static Observable<Long> interval(long interval, TimeUnit unit, Scheduler
         return create(OperationInterval.interval(interval, unit, scheduler));
     }
 
+    /**
+     * Emits one item after a given delay, and then completes.
+     * 
+     * @param interval
+     *            interval size in time units
+     * @param unit
+     *            time units to use for the interval size
+     */
+    public static Observable<Void> timer(long interval, TimeUnit unit) {
+        return create(OperationTimer.timer(interval, unit));
+    }
+
+    /**
+     * Emits one item after a given delay, and then completes.
+     * 
+     * @param interval
+     *            interval size in time units
+     * @param unit
+     *            time units to use for the interval size
+     * @param scheduler
+     *            the scheduler to use for scheduling the item
+     */
+    public static Observable<Void> timer(long interval, TimeUnit unit, Scheduler scheduler) {
+        return create(OperationTimer.timer(interval, unit, scheduler));
+    }
+
+    /**
+     * Returns an Observable that emits the results of shifting the items emitted by the source
+     * Observable by a specified delay. Errors emitted by the source Observable are not delayed.
+     * @param delay
+     *            the delay to shift the source by
+     * @param unit
+     *            the {@link TimeUnit} in which <code>period</code> is defined
+     * @return the source Observable, but shifted by the specified delay
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229810%28v=vs.103%29.aspx">MSDN: Observable.Delay</a>
+     */
+    public Observable<T> delay(long delay, TimeUnit unit) {
+        return OperationDelay.delay(this, delay, unit, Schedulers.threadPoolForComputation());
+    }
+
+    /**
+     * Returns an Observable that emits the results of shifting the items emitted by the source
+     * Observable by a specified delay. Errors emitted by the source Observable are not delayed.
+     * @param delay
+     *            the delay to shift the source by
+     * @param unit
+     *            the {@link TimeUnit} in which <code>period</code> is defined
+     * @param scheduler
+     *            the {@link Scheduler} to use for delaying
+     * @return the source Observable, but shifted by the specified delay
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229280(v=vs.103).aspx">MSDN: Observable.Delay</a>
+     */
+    public Observable<T> delay(long delay, TimeUnit unit, Scheduler scheduler) {
+        return OperationDelay.delay(this, delay, unit, scheduler);
+    }
+
     /**
      * Drops items emitted by an Observable that are followed by newer items
      * before a timeout value expires. The timer resets on each emission.
diff --git a/rxjava-core/src/main/java/rx/operators/OperationDelay.java b/rxjava-core/src/main/java/rx/operators/OperationDelay.java
@@ -0,0 +1,42 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import java.util.concurrent.TimeUnit;
+
+import rx.Observable;
+import rx.Scheduler;
+import rx.observables.ConnectableObservable;
+import rx.util.functions.Func1;
+
+public final class OperationDelay {
+
+    public static <T> Observable<T> delay(Observable<T> observable, final long delay, final TimeUnit unit, final Scheduler scheduler) {
+        // observable.map(x => Observable.timer(t).map(_ => x).startItAlreadyNow()).concat()
+        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<Void, T>() {
+                    public T call(Void ignored) {
+                        return x;
+                    }
+                }).replay();
+                co.connect();
+                return co;
+            }
+        });
+        return Observable.concat(seqs);
+    }
+}
diff --git a/rxjava-core/src/main/java/rx/operators/OperationTimer.java b/rxjava-core/src/main/java/rx/operators/OperationTimer.java
@@ -0,0 +1,74 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import java.util.concurrent.TimeUnit;
+
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Scheduler;
+import rx.Subscription;
+import rx.concurrency.Schedulers;
+import rx.subscriptions.Subscriptions;
+import rx.util.functions.Action0;
+
+public final class OperationTimer {
+
+	public static OnSubscribeFunc<Void> timer(long interval, TimeUnit unit) {
+        return timer(interval, unit, Schedulers.threadPoolForComputation());
+    }
+
+    public static OnSubscribeFunc<Void> timer(final long delay, final TimeUnit unit, final Scheduler scheduler) {
+        return new OnSubscribeFunc<Void>() {
+            @Override
+            public Subscription onSubscribe(Observer<? super Void> observer) {
+                return new Timer(delay, unit, scheduler, observer).start();
+            }
+        };
+    }
+	
+	private static class Timer {
+        private final long period;
+        private final TimeUnit unit;
+        private final Scheduler scheduler;
+        private final Observer<? super Void> observer;
+        
+        private Timer(long period, TimeUnit unit, Scheduler scheduler, Observer<? super Void> observer) {
+            this.period = period;
+            this.unit = unit;
+            this.scheduler = scheduler;
+            this.observer = observer;
+        }
+
+        public Subscription start() {
+            final Subscription s = scheduler.schedule(new Action0() {
+                @Override
+                public void call() {
+                    observer.onNext(null);
+                    observer.onCompleted();
+                }
+            }, period, unit);
+
+            return Subscriptions.create(new Action0() {
+                @Override
+                public void call() {
+                    s.unsubscribe();
+                }
+            });
+        }
+    }
+
+}
diff --git a/rxjava-core/src/test/java/rx/operators/OperationDelayTest.java b/rxjava-core/src/test/java/rx/operators/OperationDelayTest.java
@@ -0,0 +1,181 @@
+package rx.operators;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.MockitoAnnotations.initMocks;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.mockito.InOrder;
+import org.mockito.Mock;
+
+import rx.Observable;
+import rx.Observer;
+import rx.concurrency.TestScheduler;
+import rx.util.functions.Func1;
+
+public class OperationDelayTest {
+    @Mock
+    private Observer<Long> observer;
+    @Mock
+    private Observer<Long> observer2;
+
+    private TestScheduler scheduler;
+
+    @Before
+    public void before() {
+        initMocks(this);
+        scheduler = new TestScheduler();
+    }
+    
+    @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();
+    }
+    
+    // TODO activate this test once https://github.com/Netflix/RxJava/issues/552 is fixed
+    @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));
+    }
+}
