prevent repeititious emissions, add unit tests

Author: thomasnield

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

@@ -22,37 +22,38 @@
 
 /**
  * A CompositeObservable can merge multiple Observables that can be added/removed at any time,
- * affecting all Subscribers regardless of when they subscribed. This is especailly helpful for merging
+ * affecting all Subscribers regardless of when they subscribed. This is especially helpful for merging
  * multiple UI event sources.
+ *
  * @param <T>
  */
 public final class CompositeObservable<T> {
 
     private final ObservableList<Observable<T>> sources;
-    private final Observable<T> observable;
+    private final int initialCapacity;
 
     public CompositeObservable() {
         this(-1);
     }
 
     public CompositeObservable(int initialCapacity) {
+        this.initialCapacity = initialCapacity;
         sources = FXCollections.synchronizedObservableList(FXCollections.observableArrayList());
+    }
 
-        Observable<T> observable = JavaFxObservable.fromObservableList(sources)
-                .switchMap(list -> Observable.from(list).flatMap((Observable<T> obs) -> obs));
+    public Observable<T> toObservable() {
+        Observable<T> updatingSource = Observable.merge(
+                Observable.from(sources).flatMap(obs -> obs.takeWhile(v -> sources.contains(obs))),
+                JavaFxObservable.fromObservableListAdds(sources).flatMap(obs -> obs.takeWhile(v -> sources.contains(obs)))
+        );
 
         if (initialCapacity > 0) {
-            this.observable = observable.cacheWithInitialCapacity(initialCapacity);
-        }
-        else {
-            this.observable = observable;
+            return updatingSource.cacheWithInitialCapacity(initialCapacity);
+        } else {
+            return updatingSource;
         }
     }
 
-    public Observable<T> toObservable() {
-        return observable;
-    }
-
     public void add(Observable<T> observable) {
         sources.add(observable);
     }

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

@@ -28,8 +28,11 @@
 import rx.observables.JavaFxObservable;
 import rx.schedulers.JavaFxScheduler;
 import rx.schedulers.Schedulers;
+import rx.subjects.PublishSubject;
 
 import java.util.Arrays;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 
 import static org.junit.Assert.assertTrue;
@@ -293,4 +296,88 @@ public String toString() {
             e.printStackTrace();
         }
     }
+
+    @Test
+    public void testcompositeObservableInfinite() {
+
+        new JFXPanel();
+
+        CountDownLatch latch = new CountDownLatch(1);
+
+        Platform.runLater(() -> {
+            final List<String> emissions = new ArrayList<>();
+            CompositeObservable<String> compositeObservable = new CompositeObservable<>();
+
+            PublishSubject<String> source1 = PublishSubject.create();
+            PublishSubject<String> source2 = PublishSubject.create();
+            PublishSubject<String> source3 = PublishSubject.create();
+
+            compositeObservable.add(source1);
+            compositeObservable.add(source2);
+            compositeObservable.add(source3);
+
+            compositeObservable.toObservable().subscribe(emissions::add);
+
+            source1.onNext("Alpha");
+            assertTrue(emissions.get(0).equals("Alpha"));
+
+            source2.onNext("Beta");
+            assertTrue(emissions.get(1).equals("Beta"));
+
+            source3.onNext("Gamma");
+            assertTrue(emissions.get(2).equals("Gamma"));
+
+            source1.onNext("Delta");
+            assertTrue(emissions.get(3).equals("Delta"));
+
+            compositeObservable.remove(source2);
+
+            source2.onNext("Epsilon");
+            assertTrue(emissions.size() == 4);
+
+            latch.countDown();
+        });
+
+        try {
+            latch.await();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Test
+    public void testcompositeObservableFinite() {
+
+        new JFXPanel();
+
+        CountDownLatch latch = new CountDownLatch(1);
+
+        Platform.runLater(() -> {
+            final List<String> emissions = new ArrayList<>();
+            CompositeObservable<String> compositeObservable = new CompositeObservable<>();
+
+            Observable<String> source1 = Observable.just("Alpha","Beta");
+            Observable<String> source2 = Observable.just("Gamma","Delta");
+
+            compositeObservable.add(source1);
+
+            compositeObservable.toObservable().subscribe(emissions::add);
+
+            compositeObservable.add(source2);
+
+            assertTrue(emissions.size() == 4);
+
+            compositeObservable.remove(source2);
+
+            assertTrue(emissions.size() == 4);
+
+            latch.countDown();
+        });
+
+        try {
+            latch.await();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
 }