Merge pull request ReactiveX#379 from jmhofer/interval-multiple-subsc…

…ribers Make `interval` work with multiple subscribers
rickbw · Sep 13, 2013 · a73c6d9 · a73c6d9
2 parents 4f13178 + 87b0ce9
commit a73c6d9
Showing 1 changed file with 132 additions and 3 deletions.
diff --git a/rxjava-core/src/main/java/rx/operators/OperationInterval.java b/rxjava-core/src/main/java/rx/operators/OperationInterval.java
@@ -32,6 +32,7 @@
 import rx.Subscription;
 import rx.concurrency.Schedulers;
 import rx.concurrency.TestScheduler;
+import rx.observables.ConnectableObservable;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
 
@@ -45,14 +46,20 @@ public final class OperationInterval {
      * Creates an event each time interval.
      */
     public static OnSubscribeFunc<Long> interval(long interval, TimeUnit unit) {
-        return new Interval(interval, unit, Schedulers.executor(Executors.newSingleThreadScheduledExecutor()));
+        return interval(interval, unit, Schedulers.executor(Executors.newSingleThreadScheduledExecutor()));
     }
 
     /**
      * Creates an event each time interval.
      */
-    public static OnSubscribeFunc<Long> interval(long interval, TimeUnit unit, Scheduler scheduler) {
-        return new Interval(interval, unit, scheduler);
+    public static OnSubscribeFunc<Long> interval(final long interval, final TimeUnit unit, final Scheduler scheduler) {
+        // wrapped in order to work with multiple subscribers
+        return new OnSubscribeFunc<Long>() {
+            @Override
+            public Subscription onSubscribe(Observer<? super Long> observer) {
+                return new Interval(interval, unit, scheduler).onSubscribe(observer);
+            }
+        };
     }
 
     private static class Interval implements OnSubscribeFunc<Long> {
@@ -91,12 +98,14 @@ public void call() {
     public static class UnitTest {
         private TestScheduler scheduler;
         private Observer<Long> observer;
+        private Observer<Long> observer2;
 
         @Before
         @SuppressWarnings("unchecked") // due to mocking
         public void before() {
             scheduler = new TestScheduler();
             observer = mock(Observer.class);
+            observer2 = mock(Observer.class);
         }
 
         @Test
@@ -123,5 +132,125 @@ public void testInterval() {
             verify(observer, times(1)).onCompleted();
             verify(observer, never()).onError(any(Throwable.class));
         }
+
+        @Test
+        public void testWithMultipleSubscribersStartingAtSameTime() {
+            Observable<Long> w = Observable.create(OperationInterval.interval(1, TimeUnit.SECONDS, scheduler));
+            Subscription sub1 = w.subscribe(observer);
+            Subscription sub2 = w.subscribe(observer2);
+
+            verify(observer, never()).onNext(anyLong());
+            verify(observer2, never()).onNext(anyLong());
+
+            scheduler.advanceTimeTo(2, TimeUnit.SECONDS);
+
+            InOrder inOrder1 = inOrder(observer);
+            InOrder inOrder2 = inOrder(observer2);
+
+            inOrder1.verify(observer, times(1)).onNext(0L);
+            inOrder1.verify(observer, times(1)).onNext(1L);
+            inOrder1.verify(observer, never()).onNext(2L);
+            verify(observer, never()).onCompleted();
+            verify(observer, never()).onError(any(Throwable.class));
+
+            inOrder2.verify(observer2, times(1)).onNext(0L);
+            inOrder2.verify(observer2, times(1)).onNext(1L);
+            inOrder2.verify(observer2, never()).onNext(2L);
+            verify(observer2, never()).onCompleted();
+            verify(observer2, never()).onError(any(Throwable.class));
+
+            sub1.unsubscribe();
+            sub2.unsubscribe();
+            scheduler.advanceTimeTo(4, TimeUnit.SECONDS);
+
+            verify(observer, never()).onNext(2L);
+            verify(observer, times(1)).onCompleted();
+            verify(observer, never()).onError(any(Throwable.class));
+
+            verify(observer2, never()).onNext(2L);
+            verify(observer2, times(1)).onCompleted();
+            verify(observer2, never()).onError(any(Throwable.class));
+        }
+
+        @Test
+        public void testWithMultipleStaggeredSubscribers() {
+            Observable<Long> w = Observable.create(OperationInterval.interval(1, TimeUnit.SECONDS, scheduler));
+            Subscription sub1 = w.subscribe(observer);
+
+            verify(observer, never()).onNext(anyLong());
+
+            scheduler.advanceTimeTo(2, TimeUnit.SECONDS);
+            Subscription sub2 = w.subscribe(observer2);
+
+            InOrder inOrder1 = inOrder(observer);
+            inOrder1.verify(observer, times(1)).onNext(0L);
+            inOrder1.verify(observer, times(1)).onNext(1L);
+            inOrder1.verify(observer, never()).onNext(2L);
+
+            verify(observer, never()).onCompleted();
+            verify(observer, never()).onError(any(Throwable.class));
+            verify(observer2, never()).onNext(anyLong());
+
+            scheduler.advanceTimeTo(4, TimeUnit.SECONDS);
+
+            inOrder1.verify(observer, times(1)).onNext(2L);
+            inOrder1.verify(observer, times(1)).onNext(3L);
+
+            InOrder inOrder2 = inOrder(observer2);
+            inOrder2.verify(observer2, times(1)).onNext(0L);
+            inOrder2.verify(observer2, times(1)).onNext(1L);
+
+            sub1.unsubscribe();
+            sub2.unsubscribe();
+
+            inOrder1.verify(observer, never()).onNext(anyLong());
+            inOrder1.verify(observer, times(1)).onCompleted();
+            verify(observer, never()).onError(any(Throwable.class));
+
+            inOrder2.verify(observer2, never()).onNext(anyLong());
+            inOrder2.verify(observer2, times(1)).onCompleted();
+            verify(observer2, never()).onError(any(Throwable.class));
+        }
+
+        @Test
+        public void testWithMultipleStaggeredSubscribersAndPublish() {
+            ConnectableObservable<Long> w = Observable.create(OperationInterval.interval(1, TimeUnit.SECONDS, scheduler)).publish();
+            Subscription sub1 = w.subscribe(observer);
+            w.connect();
+
+            verify(observer, never()).onNext(anyLong());
+
+            scheduler.advanceTimeTo(2, TimeUnit.SECONDS);
+            Subscription sub2 = w.subscribe(observer2);
+
+            InOrder inOrder1 = inOrder(observer);
+            inOrder1.verify(observer, times(1)).onNext(0L);
+            inOrder1.verify(observer, times(1)).onNext(1L);
+            inOrder1.verify(observer, never()).onNext(2L);
+
+            verify(observer, never()).onCompleted();
+            verify(observer, never()).onError(any(Throwable.class));
+            verify(observer2, never()).onNext(anyLong());
+
+            scheduler.advanceTimeTo(4, TimeUnit.SECONDS);
+
+            inOrder1.verify(observer, times(1)).onNext(2L);
+            inOrder1.verify(observer, times(1)).onNext(3L);
+
+            InOrder inOrder2 = inOrder(observer2);
+            inOrder2.verify(observer2, times(1)).onNext(2L);
+            inOrder2.verify(observer2, times(1)).onNext(3L);
+
+            sub1.unsubscribe();
+            sub2.unsubscribe();
+
+            inOrder1.verify(observer, never()).onNext(anyLong());
+            inOrder1.verify(observer, never()).onCompleted();
+            verify(observer, never()).onError(any(Throwable.class));
+
+            inOrder2.verify(observer2, never()).onNext(anyLong());
+            inOrder2.verify(observer2, never()).onCompleted();
+            verify(observer2, never()).onError(any(Throwable.class));
+        }
     }
 }