Operation Timer

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

@@ -91,6 +91,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;
@@ -5232,7 +5233,67 @@ public Boolean call(T t) {
     public Observable<T> ignoreElements() {
         return filter(alwaysFalse());
     }
-
+
+    /**
+     * Return an Observable which emits a single 0L after the specified
+     * delay time and completes.
+     * 
+     * @param delay the delay time
+     * @param unit the delay unit
+     * @return an Observable which emits a single 0L after the specified delay time and completes.
+     * 
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh212050.aspx'>MSDN: Observable.Timer</a>
+     */
+    public static Observable<Long> timer(long delay, TimeUnit unit) {
+        return timer(delay, unit, Schedulers.threadPoolForComputation());
+    }
+
+    /**
+     * Return an Observable which emits a single 0L after the specified
+     * delay time and completes while running on the specified scheduler.
+     * 
+     * @param delay the delay time
+     * @param unit the delay unit
+     * @param scheduler the scheduler where the wait and emit should happen
+     * @return an Observable which emits a single 0L after the specified delay time and completes.
+     * 
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229503.aspx'>MSDN: Observable.Timer</a>
+     */
+    public static Observable<Long> timer(long delay, TimeUnit unit, Scheduler scheduler) {
+        return create(new OperationTimer.TimerOnce(delay, unit, scheduler));
+    }
+
+    /**
+     * Return an Observable which emits a 0L after the initialDelay and ever increasing
+     * numbers after each period.
+     * 
+     * @param initialDelay the initial delay time to wait before emitting the first value of 0L
+     * @param period the time period after emitting the subsequent numbers
+     * @param unit the time unit for both <code>initialDelay</code> and <code>period</code>
+     * @return an Observable which emits a 0L after the initialDelay and ever increasing
+     *         numbers after each period
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229435.aspx'>MSDN: Observable.Timer</a>
+     */
+    public static Observable<Long> timer(long initialDelay, long period, TimeUnit unit) {
+        return timer(initialDelay, period, unit, Schedulers.threadPoolForComputation());
+    }
+
+    /**
+     * Return an Observable which emits a 0L after the initialDelay and ever increasing
+     * numbers after each period while running on the given scheduler.
+     * 
+     * @param initialDelay the initial delay time to wait before emitting the first value of 0L
+     * @param period the time period after emitting the subsequent numbers
+     * @param unit the time unit for both <code>initialDelay</code> and <code>period</code>
+     * @param scheduler the scheduler where the waiting happens and value emissions run.
+     * @return an Observable which emits a 0L after the initialDelay and ever increasing
+     *         numbers after each period while running on the given scheduler
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229652.aspx'>MSDN: Observable.Timer</a>
+     */
+    public static Observable<Long> timer(long initialDelay, long period, TimeUnit unit, Scheduler scheduler) {
+        return create(new OperationTimer.TimerPeriodically(initialDelay, period, unit, scheduler));
+    }
+
     /**
      * Applies a timeout policy for each element in the observable sequence,
      * using the specified scheduler to run timeout timers. If the next element

rxjava-core/src/main/java/rx/operators/OperationTimer.java

@@ -0,0 +1,86 @@
+ /**
+  * 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.util.functions.Action0;
+
+/**
+ * Operation Timer with several overloads.
+ *
+ * @see <a href='http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.timer.aspx'>MSDN Observable.Timer</a>
+ */
+public final class OperationTimer {
+    private OperationTimer() { throw new IllegalStateException("No instances!"); }
+
+    /**
+     * Emit a single 0L after the specified time elapses.
+     */
+    public static class TimerOnce implements OnSubscribeFunc<Long> {
+        private final Scheduler scheduler;
+        private final long dueTime;
+        private final TimeUnit dueUnit;
+        public TimerOnce(long dueTime, TimeUnit unit, Scheduler scheduler) {
+            this.scheduler = scheduler;
+            this.dueTime = dueTime;
+            this.dueUnit = unit;
+        }
+
+        @Override
+        public Subscription onSubscribe(final Observer<? super Long> t1) {
+            return scheduler.schedule(new Action0() {
+                @Override
+                public void call() {
+                    t1.onNext(0L);
+                    t1.onCompleted();
+                }
+
+            }, dueTime, dueUnit);
+        }
+    }
+    /**
+     * Emit 0L after the initial period and ever increasing number after each period.
+     */
+    public static class TimerPeriodically implements OnSubscribeFunc<Long> {
+        private final Scheduler scheduler;
+        private final long initialDelay;
+        private final long period;
+        private final TimeUnit unit;
+        public TimerPeriodically(long initialDelay, long period, TimeUnit unit, Scheduler scheduler) {
+            this.scheduler = scheduler;
+            this.initialDelay = initialDelay;
+            this.period = period;
+            this.unit = unit;
+        }
+
+        @Override
+        public Subscription onSubscribe(final Observer<? super Long> t1) {
+            return scheduler.schedulePeriodically(new Action0() {
+                long count;
+                @Override
+                public void call() {
+                    t1.onNext(count++);
+                }
+            },
+            initialDelay, period, unit
+            );
+        }
+    }
+}

rxjava-core/src/test/java/rx/operators/OperationTimerTest.java

@@ -0,0 +1,72 @@
+/**
+ * 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 org.junit.Before;
+import org.junit.Test;
+import org.mockito.InOrder;
+import org.mockito.Mock;
+import static org.mockito.Mockito.*;
+import org.mockito.MockitoAnnotations;
+import rx.Observable;
+import rx.Observer;
+import rx.Subscription;
+import rx.concurrency.TestScheduler;
+
+public class OperationTimerTest {
+    @Mock
+    Observer<Object> observer;
+    TestScheduler s;
+    @Before
+    public void before() {
+        MockitoAnnotations.initMocks(this);
+        s = new TestScheduler();
+    }
+    @Test
+    public void testTimerOnce() {
+        Observable.timer(100, TimeUnit.MILLISECONDS, s).subscribe(observer);
+        s.advanceTimeBy(100, TimeUnit.MILLISECONDS);
+
+        verify(observer, times(1)).onNext(0L);
+        verify(observer, times(1)).onCompleted();
+        verify(observer, never()).onError(any(Throwable.class));
+    }
+    @Test
+    public void testTimerPeriodically() {
+        Subscription c = Observable.timer(100, 100, TimeUnit.MILLISECONDS, s).subscribe(observer);
+        s.advanceTimeBy(100, TimeUnit.MILLISECONDS);
+
+        InOrder inOrder = inOrder(observer);
+        inOrder.verify(observer, times(1)).onNext(0L);
+
+        s.advanceTimeBy(100, TimeUnit.MILLISECONDS);
+        inOrder.verify(observer, times(1)).onNext(1L);
+
+        s.advanceTimeBy(100, TimeUnit.MILLISECONDS);
+        inOrder.verify(observer, times(1)).onNext(2L);
+
+        s.advanceTimeBy(100, TimeUnit.MILLISECONDS);
+        inOrder.verify(observer, times(1)).onNext(3L);
+
+        c.unsubscribe();
+        s.advanceTimeBy(100, TimeUnit.MILLISECONDS);
+        inOrder.verify(observer, never()).onNext(any());
+
+        verify(observer, never()).onCompleted();
+        verify(observer, never()).onError(any(Throwable.class));
+    }
+}