Implement JavaFxObservable#interval()

Author: thomasnield

src/main/java/rx/javafx/sources/TimerSource.java

@@ -0,0 +1,27 @@
+package rx.javafx.sources;
+
+import javafx.animation.Animation;
+import javafx.animation.KeyFrame;
+import javafx.animation.Timeline;
+import javafx.util.Duration;
+import rx.Observable;
+import rx.subscriptions.JavaFxSubscriptions;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+public final class TimerSource {
+    private TimerSource() {
+    }
+
+
+    public static <T> Observable<Long> interval(final Duration duration) {
+        return Observable.create(sub -> {
+            final AtomicLong value = new AtomicLong(0);
+            Timeline timeline = new Timeline(new KeyFrame(duration, ae -> sub.onNext(value.getAndIncrement())));
+            timeline.setCycleCount(Animation.INDEFINITE);
+            timeline.play();
+
+            sub.add(JavaFxSubscriptions.unsubscribeInEventDispatchThread(timeline::stop));
+        });
+    }
+}

src/main/java/rx/observables/JavaFxObservable.java

@@ -16,6 +16,9 @@
 package rx.observables;
 
 
+import javafx.animation.Animation;
+import javafx.animation.KeyFrame;
+import javafx.animation.Timeline;
 import javafx.beans.value.ObservableValue;
 import javafx.collections.ObservableList;
 import javafx.collections.ObservableMap;
@@ -29,9 +32,12 @@
 import javafx.scene.control.MenuItem;
 import javafx.stage.Window;
 import javafx.stage.WindowEvent;
+import javafx.util.Duration;
 import rx.Observable;
 import rx.functions.Func1;
+import java.util.concurrent.atomic.AtomicLong;
 import rx.javafx.sources.*;
+import rx.subscriptions.JavaFxSubscriptions;
 
 import java.util.Map;
 
@@ -281,4 +287,11 @@ public static <T> Observable<T> fromObservableSetRemovals(final ObservableSet<T>
     public static <T> Observable<SetChange<T>> fromObservableSetChanges(final ObservableSet<T> source) {
         return ObservableSetSource.fromObservableSetChanges(source);
     }
+
+    /**
+     * Returns an Observable that emits a 0L  and ever increasing numbers after each duration of time thereafter
+     */
+    public static <T> Observable<Long> interval(final Duration duration) {
+       return TimerSource.interval(duration);
+    }
 }

src/test/java/rx/javafx/sources/JavaFxObservableTest.java

@@ -22,6 +22,7 @@
 import javafx.collections.FXCollections;
 import javafx.collections.ObservableList;
 import javafx.embed.swing.JFXPanel;
+import javafx.util.Duration;
 import org.junit.Test;
 import rx.Observable;
 import rx.observables.JavaFxObservable;
@@ -35,6 +36,22 @@
 
 public final class JavaFxObservableTest {
 
+    @Test
+    public void testIntervalSource() {
+        new JFXPanel();
+
+        final CountDownLatch latch = new CountDownLatch(5);
+
+        JavaFxObservable.interval(Duration.millis(1000)).take(5)
+                .subscribe(v -> latch.countDown());
+
+        try {
+            latch.await();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     @Test
     public void testRxObservableListAdds() {
         new JFXPanel();