@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)); }
public static OnSubscribeFunc<Long> interval(long interval, TimeUnit unit) { return interval(interval, unit, Schedulers.computation()); }
OperationInterval { public static OnSubscribeFunc<Long> interval(long interval, TimeUnit unit) { return interval(interval, unit, Schedulers.computation()); } }
OperationInterval { public static OnSubscribeFunc<Long> interval(long interval, TimeUnit unit) { return interval(interval, unit, Schedulers.computation()); }  }
OperationInterval { public static OnSubscribeFunc<Long> interval(long interval, TimeUnit unit) { return interval(interval, unit, Schedulers.computation()); }  static OnSubscribeFunc<Long> interval(long interval, TimeUnit unit); static OnSubscribeFunc<Long> interval(final long interval, final TimeUnit unit, final Scheduler scheduler); }
OperationInterval { public static OnSubscribeFunc<Long> interval(long interval, TimeUnit unit) { return interval(interval, unit, Schedulers.computation()); }  static OnSubscribeFunc<Long> interval(long interval, TimeUnit unit); static OnSubscribeFunc<Long> interval(final long interval, final TimeUnit unit, final Scheduler scheduler);  }