Implemented the delay operator

rxjava-core/src/main/java/rx/Observable.java

@@ -20,6 +20,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
+import java.util.Date;
 import java.util.List;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -35,6 +36,7 @@
 import rx.operators.OperationCombineLatest;
 import rx.operators.OperationConcat;
 import rx.operators.OperationDefer;
+import rx.operators.OperationDelay;
 import rx.operators.OperationDematerialize;
 import rx.operators.OperationDistinctUntilChanged;
 import rx.operators.OperationDistinct;
@@ -3551,6 +3553,64 @@ public Observable<T> scan(Func2<T, T, T> accumulator) {
         return create(OperationScan.scan(this, accumulator));
     }
 
+    /**
+     * Returns an Observable that emits the results of shifting the items emitted by the source
+     * Observable by a specified delay. Only 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 create(OperationDelay.delay(this, delay, unit));
+    }
+
+    /**
+     * Returns an Observable that emits the results of shifting the items emitted by the source
+     * Observable by a specified delay. Only 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 create(OperationDelay.delay(this, delay, unit, scheduler));
+    }
+
+    /**
+     * Returns an Observable that emits the results of shifting the items emitted by the source
+     * Observable by a delay specified by the due time at which to begin emitting. 
+     * Only errors emitted by the source Observable are not delayed.
+     * @param dueTime
+     *            the due time at which to start emitting
+     * @return the source Observable, but shifted by the specified delay
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229677(v=vs.103).aspx">MSDN: Observable.Delay</a>
+     */
+    public Observable<T> delay(Date dueTime) {
+        return create(OperationDelay.delay(this, dueTime));
+    }
+
+    /**
+     * Returns an Observable that emits the results of shifting the items emitted by the source
+     * Observable by a delay specified by the due time at which to begin emitting. 
+     * Only errors emitted by the source Observable are not delayed.
+     * @param dueTime
+     *            the due time at which to start emitting
+     * @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/hh229250(v=vs.103).aspx">MSDN: Observable.Delay</a>
+     */
+    public Observable<T> delay(Date dueTime, Scheduler scheduler) {
+        return create(OperationDelay.delay(this, dueTime, scheduler));
+    }
+
     /**
      * Returns an Observable that emits the results of sampling the items emitted by the source
      * Observable at a specified time interval.

rxjava-core/src/main/java/rx/concurrency/DelayedScheduler.java

@@ -0,0 +1,114 @@
+/**
+ * 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.concurrency;
+
+import static org.mockito.Mockito.*;
+import static org.mockito.MockitoAnnotations.initMocks;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.InOrder;
+import org.mockito.Mock;
+
+import rx.Scheduler;
+import rx.Subscription;
+import rx.util.functions.Action0;
+import rx.util.functions.Func2;
+
+/**
+ * Scheduler that delays the underlying scheduler by a fixed time delay.
+ */
+public class DelayedScheduler extends Scheduler {
+    private final Scheduler underlying;
+    private final long delay;
+    private final TimeUnit unit;
+
+    public DelayedScheduler(Scheduler underlying, long delay, TimeUnit unit) {
+        this.underlying = underlying;
+        this.delay = delay;
+        this.unit = unit;
+    }
+
+    @Override
+    public <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action) {
+        return underlying.schedule(state, action, delay, unit);
+    }
+
+    @Override
+    public <T> Subscription schedule(T state, Func2<? super Scheduler, ? super T, ? extends Subscription> action, long delay, TimeUnit unit) {
+        long newDelay = unit.toNanos(delay) + this.unit.toNanos(this.delay); 
+        return underlying.schedule(state, action, newDelay, TimeUnit.NANOSECONDS);
+    }
+
+    public static class UnitTest {
+        @Mock 
+        Action0 action;
+
+        private TestScheduler scheduler = new TestScheduler();
+
+        @Before
+        public void before() {
+            initMocks(this);
+        }
+
+        @Test
+        public void testNotDelayingAnAction() {
+            Scheduler delayed = new DelayedScheduler(scheduler, 0, TimeUnit.SECONDS);
+            delayed.schedule(action);
+            delayed.schedule(action, 1L, TimeUnit.SECONDS);
+
+            InOrder inOrder = inOrder(action);
+
+            scheduler.triggerActions();
+            inOrder.verify(action, times(1)).call();
+
+            scheduler.advanceTimeTo(999L, TimeUnit.MILLISECONDS);
+            inOrder.verify(action, never()).call();
+
+            scheduler.advanceTimeTo(1L, TimeUnit.SECONDS);
+            inOrder.verify(action, times(1)).call();
+
+            scheduler.advanceTimeTo(5L, TimeUnit.SECONDS);
+            inOrder.verify(action, never()).call();
+        }
+
+        @Test
+        public void testdelayingAnAction() {
+            Scheduler delayed = new DelayedScheduler(scheduler, 500, TimeUnit.MILLISECONDS);
+            delayed.schedule(action);
+            delayed.schedule(action, 1L, TimeUnit.SECONDS);
+
+            InOrder inOrder = inOrder(action);
+
+            scheduler.advanceTimeTo(499L, TimeUnit.MILLISECONDS);
+            inOrder.verify(action, never()).call();
+
+            scheduler.advanceTimeTo(500L, TimeUnit.MILLISECONDS);
+            inOrder.verify(action, times(1)).call();
+
+            scheduler.advanceTimeTo(1499L, TimeUnit.MILLISECONDS);
+            inOrder.verify(action, never()).call();
+
+            scheduler.advanceTimeTo(1500L, TimeUnit.MILLISECONDS);
+            inOrder.verify(action, times(1)).call();
+
+            scheduler.advanceTimeTo(5L, TimeUnit.SECONDS);
+            inOrder.verify(action, never()).call();
+        }
+    }
+}