From 1515b01d2154a664fa1c3f55f4c6a4ed618430ea Mon Sep 17 00:00:00 2001
From: Tomas Mikula <tomas.mikula@gmail.com>
Date: Tue, 17 Jun 2014 05:02:02 +0200
Subject: [PATCH] Add sizeOf(ObservableCollection) factory methods.

---
 .../main/java/org/reactfx/EventStreams.java   | 44 +++++++++++++++++++
 .../src/test/java/org/reactfx/SizeOfTest.java | 36 +++++++++++++++
 2 files changed, 80 insertions(+)
 create mode 100644 reactfx/src/test/java/org/reactfx/SizeOfTest.java

diff --git a/reactfx/src/main/java/org/reactfx/EventStreams.java b/reactfx/src/main/java/org/reactfx/EventStreams.java
index 26bc6b9..ce0d7b1 100644
--- a/reactfx/src/main/java/org/reactfx/EventStreams.java
+++ b/reactfx/src/main/java/org/reactfx/EventStreams.java
@@ -1,10 +1,12 @@
 package org.reactfx;
 
 import java.time.Duration;
+import java.util.Collection;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 import javafx.beans.InvalidationListener;
 import javafx.beans.Observable;
@@ -142,6 +144,48 @@ protected Subscription subscribeToInputs() {
         };
     }
 
+    public static <C extends Collection<?> & Observable> EventStream<Integer> sizeOf(C collection) {
+        return create(() -> collection.size(), collection);
+    }
+
+    public static EventStream<Integer> sizeOf(ObservableMap<?, ?> map) {
+        return create(() -> map.size(), map);
+    }
+
+    private static <T> EventStream<T> create(Supplier<? extends T> computeValue, Observable... dependencies) {
+        return new LazilyBoundStream<T>() {
+            private T previousValue;
+
+            @Override
+            protected Subscription subscribeToInputs() {
+                InvalidationListener listener = obs -> {
+                    T value = computeValue.get();
+                    if(value != previousValue) {
+                        previousValue = value;
+                        emit(value);
+                    }
+                };
+                for(Observable dep: dependencies) {
+                    dep.addListener(listener);
+                }
+
+                return () -> {
+                    for(Observable dep: dependencies) {
+                        dep.removeListener(listener);
+                    }
+                };
+            }
+
+            @Override
+            protected void newSubscriber(Consumer<? super T> subscriber) {
+                if(!isBound()) { // this is the first subscriber
+                    previousValue = computeValue.get();
+                }
+                subscriber.accept(previousValue);
+            }
+        };
+    }
+
     public static <T extends Event> EventStream<T> eventsOf(Node node, EventType<T> eventType) {
         return new LazilyBoundStream<T>() {
             @Override
diff --git a/reactfx/src/test/java/org/reactfx/SizeOfTest.java b/reactfx/src/test/java/org/reactfx/SizeOfTest.java
new file mode 100644
index 0000000..af438af
--- /dev/null
+++ b/reactfx/src/test/java/org/reactfx/SizeOfTest.java
@@ -0,0 +1,36 @@
+package org.reactfx;
+
+import static org.junit.Assert.*;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import javafx.collections.FXCollections;
+import javafx.collections.ObservableList;
+
+import org.junit.Test;
+
+public class SizeOfTest {
+
+    @Test
+    public void test() {
+        ObservableList<Integer> list = FXCollections.observableArrayList();
+        EventStream<Integer> size = EventStreams.sizeOf(list);
+        List<Integer> sizes = new ArrayList<>();
+        Subscription sub = size.subscribe(sizes::add);
+        list.add(1);
+        list.addAll(2, 3, 4);
+        assertEquals(Arrays.asList(0, 1, 4), sizes);
+
+        sub.unsubscribe();
+        sizes.clear();
+        list.addAll(5, 6);
+        assertEquals(Arrays.asList(), sizes);
+
+        size.subscribe(sizes::add);
+        list.addAll(7, 8);
+        assertEquals(Arrays.asList(6, 8), sizes);
+    }
+
+}