OperationTimeInterval { public static <T> OnSubscribeFunc<TimeInterval<T>> timeInterval( Observable<? extends T> source) { return timeInterval(source, Schedulers.immediate()); }  static OnSubscribeFunc<TimeInterval<T>> timeInterval(
            Observable<? extends T> source); static OnSubscribeFunc<TimeInterval<T>> timeInterval(
            final Observable<? extends T> source, final Scheduler scheduler);  }
@Test public void testTimeInterval() { InOrder inOrder = inOrder(observer); observable.subscribe(observer); testScheduler.advanceTimeBy(1000, TIME_UNIT); subject.onNext(1); testScheduler.advanceTimeBy(2000, TIME_UNIT); subject.onNext(2); testScheduler.advanceTimeBy(3000, TIME_UNIT); subject.onNext(3); subject.onCompleted(); inOrder.verify(observer, times(1)).onNext( new TimeInterval<Integer>(1000, 1)); inOrder.verify(observer, times(1)).onNext( new TimeInterval<Integer>(2000, 2)); inOrder.verify(observer, times(1)).onNext( new TimeInterval<Integer>(3000, 3)); inOrder.verify(observer, times(1)).onCompleted(); inOrder.verifyNoMoreInteractions(); }