@Test public void normal() { TestSubscriber<Long> ts = new TestSubscriber<>(Long.MAX_VALUE); Multi.interval(1, TimeUnit.MILLISECONDS, executor) .subscribe(ts); ts.awaitCount(10) .cancel() .assertNotTerminated(); }
@Override public void subscribe(Flow.Subscriber<? super Long> subscriber) { IntervalSubscription subscription = new IntervalSubscription(subscriber); subscriber.onSubscribe(subscription); subscription.setFuture(executor.scheduleAtFixedRate(subscription, initialDelay, period, unit)); }
MultiInterval implements Multi<Long> { @Override public void subscribe(Flow.Subscriber<? super Long> subscriber) { IntervalSubscription subscription = new IntervalSubscription(subscriber); subscriber.onSubscribe(subscription); subscription.setFuture(executor.scheduleAtFixedRate(subscription, initialDelay, period, unit)); } }
MultiInterval implements Multi<Long> { @Override public void subscribe(Flow.Subscriber<? super Long> subscriber) { IntervalSubscription subscription = new IntervalSubscription(subscriber); subscriber.onSubscribe(subscription); subscription.setFuture(executor.scheduleAtFixedRate(subscription, initialDelay, period, unit)); } MultiInterval(long initialDelay, long period, TimeUnit unit, ScheduledExecutorService executor); }
MultiInterval implements Multi<Long> { @Override public void subscribe(Flow.Subscriber<? super Long> subscriber) { IntervalSubscription subscription = new IntervalSubscription(subscriber); subscriber.onSubscribe(subscription); subscription.setFuture(executor.scheduleAtFixedRate(subscription, initialDelay, period, unit)); } MultiInterval(long initialDelay, long period, TimeUnit unit, ScheduledExecutorService executor); @Override void subscribe(Flow.Subscriber<? super Long> subscriber); }
MultiInterval implements Multi<Long> { @Override public void subscribe(Flow.Subscriber<? super Long> subscriber) { IntervalSubscription subscription = new IntervalSubscription(subscriber); subscriber.onSubscribe(subscription); subscription.setFuture(executor.scheduleAtFixedRate(subscription, initialDelay, period, unit)); } MultiInterval(long initialDelay, long period, TimeUnit unit, ScheduledExecutorService executor); @Override void subscribe(Flow.Subscriber<? super Long> subscriber);  }