Add EventStreams.ticks(Duration) factory method.

reactfx/src/main/java/org/reactfx/EventStream.java

@@ -419,7 +419,7 @@ default <U> AwaitingEventStream<U> reduceSuccessions(
             Executor eventThreadExecutor) {
 
         Function<Runnable, Timer> timerFactory =
-                action -> new ScheduledExecutorServiceTimer(
+                action -> ScheduledExecutorServiceTimer.create(
                         timeout, action, scheduler, eventThreadExecutor);
         return new SuccessionReducingStream<T, U>(
                 this, initialTransformation, reduction, timerFactory);

reactfx/src/main/java/org/reactfx/EventStreams.java

@@ -1,5 +1,8 @@
 package org.reactfx;
 
+import java.time.Duration;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Consumer;
 import java.util.stream.Stream;
 
@@ -18,6 +21,9 @@
 import javafx.event.EventType;
 import javafx.scene.Node;
 
+import org.reactfx.util.FxTimer;
+import org.reactfx.util.Timer;
+
 public class EventStreams {
 
     /**
@@ -132,6 +138,46 @@ protected Subscription subscribeToInputs() {
         };
     }
 
+    /**
+     * Returns an event stream that emits periodic <i>ticks</i>. The returned
+     * stream may only be used on the JavaFX application thread.
+     */
+    public static EventStream<?> ticks(Duration interval) {
+        return new LazilyBoundStream<Void>() {
+            private final Timer timer = FxTimer.createPeriodic(
+                    interval, () -> emit(null));
+
+            @Override
+            protected Subscription subscribeToInputs() {
+                timer.restart();
+                return timer::stop;
+            }
+        };
+    }
+
+    /**
+     * Returns an event stream that emits periodic <i>ticks</i> on the given
+     * {@code eventThreadExecutor}. The returned stream may only be used from
+     * that executor's thread.
+     * @param scheduler scheduler used to schedule periodic emissions.
+     * @param eventThreadExecutor single-thread executor used to emit the ticks.
+     */
+    public static EventStream<?> ticks(
+            Duration interval,
+            ScheduledExecutorService scheduler,
+            Executor eventThreadExecutor) {
+        return new LazilyBoundStream<Void>() {
+            private final Timer timer = ScheduledExecutorServiceTimer.createPeriodic(
+                    interval, () -> emit(null), scheduler, eventThreadExecutor);
+
+            @Override
+            protected Subscription subscribeToInputs() {
+                timer.restart();
+                return timer::stop;
+            }
+        };
+    }
+
     /**
      * Returns an event stream that emits all the events emitted from any of
      * the {@code inputs}. The event type of the returned stream is the nearest

reactfx/src/main/java/org/reactfx/ScheduledExecutorServiceTimer.java

@@ -6,21 +6,45 @@
 import java.util.concurrent.TimeUnit;
 
 import org.reactfx.util.Timer;
+import org.reactfx.util.TriFunction;
 
 class ScheduledExecutorServiceTimer implements Timer {
+
+    public static Timer create(
+            java.time.Duration timeout,
+            Runnable action,
+            ScheduledExecutorService scheduler,
+            Executor eventThreadExecutor) {
+        return new ScheduledExecutorServiceTimer(
+                timeout, action,
+                (delay, unit, cmd) -> scheduler.schedule(cmd, delay, unit),
+                eventThreadExecutor);
+    }
+
+    public static Timer createPeriodic(
+            java.time.Duration timeout,
+            Runnable action,
+            ScheduledExecutorService scheduler,
+            Executor eventThreadExecutor) {
+        return new ScheduledExecutorServiceTimer(
+                timeout, action,
+                (delay, unit, cmd) -> scheduler.scheduleAtFixedRate(cmd, delay, delay, unit),
+                eventThreadExecutor);
+    }
+
     private final long timeout;
     private final TimeUnit unit;
     private final Runnable action;
-    private final ScheduledExecutorService scheduler;
+    private final TriFunction<Long, TimeUnit, Runnable, ScheduledFuture<?>> scheduler;
     private final Executor eventThreadExecutor;
 
     private ScheduledFuture<?> pendingTimer = null;
     private long seq = 0;
 
-    public ScheduledExecutorServiceTimer(
+    private ScheduledExecutorServiceTimer(
             java.time.Duration timeout,
             Runnable action,
-            ScheduledExecutorService scheduler,
+            TriFunction<Long, TimeUnit, Runnable, ScheduledFuture<?>> scheduler,
             Executor eventThreadExecutor) {
 
         this.timeout = timeout.toNanos();
@@ -34,13 +58,13 @@ public ScheduledExecutorServiceTimer(
     public final void restart() {
         stop();
         long expected = seq;
-        pendingTimer = scheduler.schedule(
-                () -> eventThreadExecutor.execute(() -> {
-                    if(seq == expected) {
-                        action.run();
-                    }
-                }),
-                timeout, unit);
+        pendingTimer = scheduler.apply(timeout, unit, () -> {
+            eventThreadExecutor.execute(() -> {
+                if(seq == expected) {
+                    action.run();
+                }
+            });
+        });
     }
 
     @Override

reactfx/src/test/java/org/reactfx/EventCounter.java

@@ -0,0 +1,22 @@
+package org.reactfx;
+
+import java.util.function.Consumer;
+
+class EventCounter implements Consumer<Object> {
+    private int count = 0;
+
+    @Override
+    public void accept(Object o) {
+        ++count;
+    }
+
+    public int get() {
+        return count;
+    }
+
+    public int getAndReset() {
+        int res = count;
+        count = 0;
+        return res;
+    }
+}

reactfx/src/test/java/org/reactfx/TicksTest.java

@@ -0,0 +1,64 @@
+package org.reactfx;
+
+import static org.junit.Assert.*;
+
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+import javafx.application.Platform;
+import javafx.embed.swing.JFXPanel;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.reactfx.util.FxTimer;
+
+
+public class TicksTest {
+    private ScheduledExecutorService scheduler;
+
+    @Before
+    public void setUp() throws Exception {
+        scheduler = Executors.newSingleThreadScheduledExecutor();
+        new JFXPanel(); // initializes JavaFX toolkit
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        scheduler.shutdown();
+    }
+
+    @Test
+    public void fxTest() throws InterruptedException, ExecutionException {
+        CompletableFuture<Integer> nTicks = new CompletableFuture<>();
+        Platform.runLater(() -> {
+            EventCounter counter = new EventCounter();
+            Subscription sub = EventStreams.ticks(Duration.ofMillis(100)).subscribe(counter);
+            FxTimer.runLater(Duration.ofMillis(350), sub::unsubscribe); // stop after 3 ticks
+            // wait a little more to test that no more than 3 ticks arrive anyway
+            FxTimer.runLater(Duration.ofMillis(550), () -> nTicks.complete(counter.get()));
+        });
+        assertEquals(3, nTicks.get().intValue());
+    }
+
+    @Test
+    public void executorTest() throws InterruptedException, ExecutionException {
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+
+        CompletableFuture<Integer> nTicks = new CompletableFuture<>();
+        executor.execute(() -> {
+            EventCounter counter = new EventCounter();
+            Subscription sub = EventStreams.ticks(Duration.ofMillis(100), scheduler, executor).subscribe(counter);
+            ScheduledExecutorServiceTimer.create(Duration.ofMillis(350), sub::unsubscribe, scheduler, executor).restart(); // stop after 3 ticks
+            // wait a little more to test that no more than 3 ticks arrive anyway
+            ScheduledExecutorServiceTimer.create(Duration.ofMillis(550), () -> nTicks.complete(counter.get()), scheduler, executor).restart();
+        });
+        assertEquals(3, nTicks.get().intValue());
+
+        executor.shutdown();
+    }
+}