Merge pull request #2975 from abersnaze/interval

Deprecate and rename two timer overloads to interval

Author: akarnokd

src/main/java/rx/Observable.java

@@ -1225,7 +1225,7 @@ public final static <T> Observable<T> from(T[] array) {
      * @see <a href="http://reactivex.io/documentation/operators/interval.html">ReactiveX operators documentation: Interval</a>
      */
     public final static Observable<Long> interval(long interval, TimeUnit unit) {
-        return interval(interval, unit, Schedulers.computation());
+        return interval(interval, interval, unit, Schedulers.computation());
     }
 
     /**
@@ -1248,7 +1248,65 @@ public final static Observable<Long> interval(long interval, TimeUnit unit) {
      * @see <a href="http://reactivex.io/documentation/operators/interval.html">ReactiveX operators documentation: Interval</a>
      */
     public final static Observable<Long> interval(long interval, TimeUnit unit, Scheduler scheduler) {
-        return create(new OnSubscribeTimerPeriodically(interval, interval, unit, scheduler));
+        return interval(interval, interval, unit, scheduler);
+    }
+
+    /**
+     * Returns an Observable that emits a {@code 0L} after the {@code initialDelay} and ever increasing numbers
+     * after each {@code period} of time thereafter.
+     * <p>
+     * <img width="640" height="200" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/timer.p.png" alt="">
+     * <dl>
+     *  <dt><b>Backpressure Support:</b></dt>
+     *  <dd>This operator does not support backpressure as it uses time. If the downstream needs a slower rate
+     *      it should slow the timer or use something like {@link #onBackpressureDrop}.</dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code timer} operates by default on the {@code computation} {@link Scheduler}.</dd>
+     * </dl>
+     * 
+     * @param initialDelay
+     *            the initial delay time to wait before emitting the first value of 0L
+     * @param period
+     *            the period of time between emissions of the subsequent numbers
+     * @param unit
+     *            the time unit for both {@code initialDelay} and {@code period}
+     * @return an Observable that emits a 0L after the {@code initialDelay} and ever increasing numbers after
+     *         each {@code period} of time thereafter
+     * @see <a href="http://reactivex.io/documentation/operators/interval.html">ReactiveX operators documentation: Interval</a>
+     * @since 1.0.12
+     */
+    public final static Observable<Long> interval(long initialDelay, long period, TimeUnit unit) {
+        return interval(initialDelay, period, unit, Schedulers.computation());
+    }
+
+    /**
+     * Returns an Observable that emits a {@code 0L} after the {@code initialDelay} and ever increasing numbers
+     * after each {@code period} of time thereafter, on a specified {@link Scheduler}.
+     * <p>
+     * <img width="640" height="200" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/timer.ps.png" alt="">
+     * <dl>
+     *  <dt><b>Backpressure Support:</b></dt>
+     *  <dd>This operator does not support backpressure as it uses time. If the downstream needs a slower rate
+     *      it should slow the timer or use something like {@link #onBackpressureDrop}.</dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>you specify which {@link Scheduler} this operator will use</dd>
+     * </dl>
+     * 
+     * @param initialDelay
+     *            the initial delay time to wait before emitting the first value of 0L
+     * @param period
+     *            the period of time between emissions of the subsequent numbers
+     * @param unit
+     *            the time unit for both {@code initialDelay} and {@code period}
+     * @param scheduler
+     *            the Scheduler on which the waiting happens and items are emitted
+     * @return an Observable that emits a 0L after the {@code initialDelay} and ever increasing numbers after
+     *         each {@code period} of time thereafter, while running on the given Scheduler
+     * @see <a href="http://reactivex.io/documentation/operators/interval.html">ReactiveX operators documentation: Interval</a>
+     * @since 1.0.12
+     */
+    public final static Observable<Long> interval(long initialDelay, long period, TimeUnit unit, Scheduler scheduler) {
+        return create(new OnSubscribeTimerPeriodically(initialDelay, period, unit, scheduler));
     }
 
     /**
@@ -2462,9 +2520,11 @@ public final static <T> Observable<T> switchOnNext(Observable<? extends Observab
      * @return an Observable that emits a 0L after the {@code initialDelay} and ever increasing numbers after
      *         each {@code period} of time thereafter
      * @see <a href="http://reactivex.io/documentation/operators/timer.html">ReactiveX operators documentation: Timer</a>
+     * @deprecated use {@link #interval(long, long, TimeUnit)} instead
      */
+    @Deprecated
     public final static Observable<Long> timer(long initialDelay, long period, TimeUnit unit) {
-        return timer(initialDelay, period, unit, Schedulers.computation());
+        return interval(initialDelay, period, unit, Schedulers.computation());
     }
 
     /**
@@ -2491,9 +2551,11 @@ public final static Observable<Long> timer(long initialDelay, long period, TimeU
      * @return an Observable that emits a 0L after the {@code initialDelay} and ever increasing numbers after
      *         each {@code period} of time thereafter, while running on the given Scheduler
      * @see <a href="http://reactivex.io/documentation/operators/timer.html">ReactiveX operators documentation: Timer</a>
+     * @deprecated use {@link #interval(long, long, TimeUnit, Scheduler)} instead
      */
+    @Deprecated
     public final static Observable<Long> timer(long initialDelay, long period, TimeUnit unit, Scheduler scheduler) {
-        return create(new OnSubscribeTimerPeriodically(initialDelay, period, unit, scheduler));
+        return interval(initialDelay, period, unit, scheduler);
     }
 
     /**

src/perf/java/rx/operators/OperatorSerializePerf.java

@@ -90,7 +90,7 @@ public int getSize() {
         public void setup(Blackhole bh) {
             super.setup(bh);
 
-            interval = Observable.timer(0, 1, TimeUnit.MILLISECONDS).take(size).map(this);
+            interval = Observable.interval(0, 1, TimeUnit.MILLISECONDS).take(size).map(this);
         }
         @Override
         public Integer call(Long t1) {

src/test/java/rx/internal/operators/OnSubscribeCacheTest.java

@@ -114,7 +114,7 @@ public Integer call(Long t1) {
 
         Observable<Integer> source2 = source1
                 .repeat(4)
-                .zipWith(Observable.timer(0, 10, TimeUnit.MILLISECONDS, Schedulers.newThread()), new Func2<Integer, Long, Integer>() {
+                .zipWith(Observable.interval(0, 10, TimeUnit.MILLISECONDS, Schedulers.newThread()), new Func2<Integer, Long, Integer>() {
                     @Override
                     public Integer call(Integer t1, Long t2) {
                         return t1;

src/test/java/rx/internal/operators/OnSubscribeCombineLatestTest.java

@@ -820,7 +820,7 @@ public void testWithCombineLatestIssue1717() throws InterruptedException {
         final CountDownLatch latch = new CountDownLatch(1);
         final AtomicInteger count = new AtomicInteger();
         final int SIZE = 2000;
-        Observable<Long> timer = Observable.timer(0, 1, TimeUnit.MILLISECONDS)
+        Observable<Long> timer = Observable.interval(0, 1, TimeUnit.MILLISECONDS)
                 .observeOn(Schedulers.newThread())
                 .doOnEach(new Action1<Notification<? super Long>>() {
 

src/test/java/rx/internal/operators/OnSubscribeRefCountTest.java

@@ -47,7 +47,7 @@ public void setUp() {
     public void testRefCountAsync() {
         final AtomicInteger subscribeCount = new AtomicInteger();
         final AtomicInteger nextCount = new AtomicInteger();
-        Observable<Long> r = Observable.timer(0, 5, TimeUnit.MILLISECONDS)
+        Observable<Long> r = Observable.interval(0, 5, TimeUnit.MILLISECONDS)
                 .doOnSubscribe(new Action0() {
 
                     @Override
@@ -183,7 +183,7 @@ public void call(Integer l) {
     public void testRepeat() {
         final AtomicInteger subscribeCount = new AtomicInteger();
         final AtomicInteger unsubscribeCount = new AtomicInteger();
-        Observable<Long> r = Observable.timer(0, 1, TimeUnit.MILLISECONDS)
+        Observable<Long> r = Observable.interval(0, 1, TimeUnit.MILLISECONDS)
                 .doOnSubscribe(new Action0() {
 
                     @Override

src/test/java/rx/internal/operators/OnSubscribeTimerTest.java

@@ -64,7 +64,7 @@ public void testTimerOnce() {
 
     @Test
     public void testTimerPeriodically() {
-        Subscription c = Observable.timer(100, 100, TimeUnit.MILLISECONDS, scheduler).subscribe(observer);
+        Subscription c = Observable.interval(100, 100, TimeUnit.MILLISECONDS, scheduler).subscribe(observer);
         scheduler.advanceTimeBy(100, TimeUnit.MILLISECONDS);
 
         InOrder inOrder = inOrder(observer);
@@ -260,7 +260,7 @@ public void onCompleted() {
     }
     @Test
     public void testPeriodicObserverThrows() {
-        Observable<Long> source = Observable.timer(100, 100, TimeUnit.MILLISECONDS, scheduler);
+        Observable<Long> source = Observable.interval(100, 100, TimeUnit.MILLISECONDS, scheduler);
 
         InOrder inOrder = inOrder(observer);
 

src/test/java/rx/internal/operators/OperatorBufferTest.java

@@ -557,7 +557,7 @@ public void bufferWithSizeSkipTake1() {
     }
     @Test(timeout = 2000)
     public void bufferWithTimeTake1() {
-        Observable<Long> source = Observable.timer(40, 40, TimeUnit.MILLISECONDS, scheduler);
+        Observable<Long> source = Observable.interval(40, 40, TimeUnit.MILLISECONDS, scheduler);
 
         Observable<List<Long>> result = source.buffer(100, TimeUnit.MILLISECONDS, scheduler).take(1);
 
@@ -574,7 +574,7 @@ public void bufferWithTimeTake1() {
     }
     @Test(timeout = 2000)
     public void bufferWithTimeSkipTake2() {
-        Observable<Long> source = Observable.timer(40, 40, TimeUnit.MILLISECONDS, scheduler);
+        Observable<Long> source = Observable.interval(40, 40, TimeUnit.MILLISECONDS, scheduler);
 
         Observable<List<Long>> result = source.buffer(100, 60, TimeUnit.MILLISECONDS, scheduler).take(2);
 
@@ -593,8 +593,8 @@ public void bufferWithTimeSkipTake2() {
     }
     @Test(timeout = 2000)
     public void bufferWithBoundaryTake2() {
-        Observable<Long> boundary = Observable.timer(60, 60, TimeUnit.MILLISECONDS, scheduler);
-        Observable<Long> source = Observable.timer(40, 40, TimeUnit.MILLISECONDS, scheduler);
+        Observable<Long> boundary = Observable.interval(60, 60, TimeUnit.MILLISECONDS, scheduler);
+        Observable<Long> source = Observable.interval(40, 40, TimeUnit.MILLISECONDS, scheduler);
 
         Observable<List<Long>> result = source.buffer(boundary).take(2);
 
@@ -615,15 +615,15 @@ public void bufferWithBoundaryTake2() {
 
     @Test(timeout = 2000)
     public void bufferWithStartEndBoundaryTake2() {
-        Observable<Long> start = Observable.timer(61, 61, TimeUnit.MILLISECONDS, scheduler);
+        Observable<Long> start = Observable.interval(61, 61, TimeUnit.MILLISECONDS, scheduler);
         Func1<Long, Observable<Long>> end = new Func1<Long, Observable<Long>>() {
             @Override
             public Observable<Long> call(Long t1) {
-                return Observable.timer(100, 100, TimeUnit.MILLISECONDS, scheduler);
+                return Observable.interval(100, 100, TimeUnit.MILLISECONDS, scheduler);
             }
         };
 
-        Observable<Long> source = Observable.timer(40, 40, TimeUnit.MILLISECONDS, scheduler);
+        Observable<Long> source = Observable.interval(40, 40, TimeUnit.MILLISECONDS, scheduler);
 
         Observable<List<Long>> result = source.buffer(start, end).take(2);
 
@@ -693,7 +693,7 @@ public void bufferWithTimeThrows() {
     }
     @Test
     public void bufferWithTimeAndSize() {
-        Observable<Long> source = Observable.timer(30, 30, TimeUnit.MILLISECONDS, scheduler);
+        Observable<Long> source = Observable.interval(30, 30, TimeUnit.MILLISECONDS, scheduler);
 
         Observable<List<Long>> result = source.buffer(100, TimeUnit.MILLISECONDS, 2, scheduler).take(3);
 

src/test/java/rx/internal/operators/OperatorObserveOnTest.java

@@ -663,7 +663,7 @@ public void onNext(Long t) {
     @Test
     public void testHotOperatorBackpressure() {
         TestSubscriber<String> ts = new TestSubscriber<String>();
-        Observable.timer(0, 1, TimeUnit.MICROSECONDS)
+        Observable.interval(0, 1, TimeUnit.MICROSECONDS)
                 .observeOn(Schedulers.computation())
                 .map(new Func1<Long, String>() {
 
@@ -687,7 +687,7 @@ public String call(Long t1) {
 
     @Test
     public void testErrorPropagatesWhenNoOutstandingRequests() {
-        Observable<Long> timer = Observable.timer(0, 1, TimeUnit.MICROSECONDS)
+        Observable<Long> timer = Observable.interval(0, 1, TimeUnit.MICROSECONDS)
                 .doOnEach(new Action1<Notification<? super Long>>() {
 
                     @Override

src/test/java/rx/internal/operators/OperatorPublishTest.java

@@ -248,7 +248,7 @@ public void call() {
     @Test
     public void testConnectWithNoSubscriber() {
         TestScheduler scheduler = new TestScheduler();
-        ConnectableObservable<Long> co = Observable.timer(10, 10, TimeUnit.MILLISECONDS, scheduler).take(3).publish();
+        ConnectableObservable<Long> co = Observable.interval(10, 10, TimeUnit.MILLISECONDS, scheduler).take(3).publish();
         co.connect();
         // Emit 0
         scheduler.advanceTimeBy(15, TimeUnit.MILLISECONDS);

src/test/java/rx/internal/producers/ProducersTest.java

@@ -355,10 +355,10 @@ public void testObserverArbiterAsync() {
         TestScheduler test = Schedulers.test();
         @SuppressWarnings("unchecked")
         List<Observable<Long>> timers = Arrays.asList(
-            Observable.timer(100, 100, TimeUnit.MILLISECONDS, test),
-            Observable.timer(100, 100, TimeUnit.MILLISECONDS, test)
+            Observable.interval(100, 100, TimeUnit.MILLISECONDS, test),
+            Observable.interval(100, 100, TimeUnit.MILLISECONDS, test)
                 .map(plus(20)),
-            Observable.timer(100, 100, TimeUnit.MILLISECONDS, test)
+            Observable.interval(100, 100, TimeUnit.MILLISECONDS, test)
                 .map(plus(40))
         );