PeriodicGenerator { PeriodicGenerator(Observable<T> sourceObservable, long initialDelay, long interval, TimeUnit timeUnit, Scheduler scheduler) { this.sourceObservable = sourceObservable; this.scheduler = scheduler; this.initialDelayMs = timeUnit.toMillis(initialDelay); this.intervalMs = timeUnit.toMillis(interval); } PeriodicGenerator(Observable<T> sourceObservable,
                      long initialDelay,
                      long interval,
                      TimeUnit timeUnit,
                      Scheduler scheduler);   }
@Test public void testOkEmitter() throws Exception { Subscription subscription = ObservableExt.periodicGenerator( OK_SOURCE, INITIAL_DELAY_MS, INTERVAL_MS, TimeUnit.MILLISECONDS, testScheduler ).subscribe(testSubscriber); assertThat(testSubscriber.takeNext()).isNull(); expectOkEmit(INITIAL_DELAY_MS); expectOkEmit(INTERVAL_MS); expectOkEmit(INTERVAL_MS); subscription.unsubscribe(); testSubscriber.onCompleted(); }
@Test public void testFailingEmitter() throws Exception { ObservableExt.periodicGenerator( BAD_SOURCE, INITIAL_DELAY_MS, INTERVAL_MS, TimeUnit.MILLISECONDS, testScheduler ).subscribe(testSubscriber); expectBadEmit(INITIAL_DELAY_MS); }