merge changes

Author: thomasnield

README.md

@@ -55,7 +55,7 @@ $ ./gradlew build
 ## Features
 
 RxJavaFX has a lightweight set of features:
-- Factories to turn `Node` and `ObservableValue` events into an RxJava `Observable`
+- Factories to turn `Node`, `ObservableValue`, and other component events into an RxJava `Observable`
 - Factories to turn an RxJava `Observable` into a JavaFX `Binding`. 
 - A scheduler for the JavaFX dispatch thread
 
@@ -93,7 +93,7 @@ Observable<ActionEvent> menuItemEvents =
 
 There are also factories provided to convert events from a `Window` as well as a `Scene` into an `Observable`. If you would like to see factories for other components and event types, please let us know or put in a PR. 
 
-#####Emitting Mouse Movement Events
+#####Emitting Scene Events
 
 ```java
 Observable<MouseEvent> sceneMouseMovements =
@@ -121,7 +121,7 @@ Observable<String> textInputs =
 ```
 Note that many Nodes in JavaFX will have an initial value, which sometimes can be `null`, and you might consider using RxJava's `skip()` operator to ignore this initial value. 
 
-###ObservableValue Changes
+#####ObservableValue Changes
 
 For every change to an `ObservableValue`, you can emit the old value and new value as a pair. The two values will be wrapped up in a `Change` class and you can access them via `getOldVal()` and `getNewVal()`. Just call the `JavaFxObservable.fromObservableValueChanges()` factory. 
 
@@ -173,6 +173,16 @@ sub2 = textInputs.observeOn(Schedulers.computation())
         .observeOn(JavaFxScheduler.getInstance())
         .subscribe(fippedTextLabel::setText);
 ```
+
+##Differences from ReactFX
+[ReactFX](https://github.com/TomasMikula/ReactFX) is a popular API to implement reactive patterns with JavaFX using the `EventStream`. However, RxJava uses an `Observable` and the two are not (directly) compatible with each other. 
+
+Although ReactFX has some asynchronous operators like `threadBridge`, ReactFX emphasizes *synchronous* behavior. This means it encourages keeping events on the JavaFX thread. RxJavaFX, which fully embraces RxJava and *asynchronous* design, can switch between threads and schedulers with ease.  As long as subscriptions affecting the UI are observed on the JavaFX thread, you can leverage the powerful operators and libraries of RxJava safely.
+
+If you are heavily dependent on RxJava, asynchronous processing, or do not want your entire reactive codebase to be UI-focused, you will probably want to use RxJavaFX. 
+
+
+
 ##Notes for Kotlin 
 If you are building your JavaFX application with [Kotlin](https://kotlinlang.org/), this library becomes even more useful with extension functions. These extension functions exist in the [RxKotlinFX](https://github.com/thomasnield/RxKotlinFX) project, but the API is so small it is not worth publishing at the moment. But you can simply add these extension functions below to your project and utilize Kotlin's fluent style.
 
@@ -181,7 +191,7 @@ fun <T> Observable<T>.toBinding() = JavaFxSubscriber.toBinding(this)
 fun <T> Observable<T>.toBinding(errorHandler: (Throwable) -> Unit) = JavaFxSubscriber.toBinding(this,errorHandler)
 fun <T> ObservableValue<T>.toObservable() = JavaFxObservable.fromObservableValue(this)
 fun <T> ObservableValue<T>.toObservableChanges() = JavaFxObservable.fromObservableValueChanges(this)
-fun <T : Event> Node.toNodeEvents(eventType: EventType<T>) = JavaFxObservable.fromNodeEvents(this, eventType)
+fun <T: Event> Node.toNodeEvents(eventType: EventType<T>) = JavaFxObservable.fromNodeEvents(this, eventType)
 fun <T> Observable<T>.observeOnFx() = this.observeOn(JavaFxScheduler.getInstance())
 ```
 This allows you to better use Kotlin's features to interop JavaFX and RxJava much more cleanly.
@@ -194,12 +204,6 @@ val lengthBinding = textInputs.map { it.length }.toBinding()
 
 If you are doing JavaFX and Kotlin development, definitely check out [TornadoFX](https://github.com/edvin/tornadofx) as well to utilize type-safe builders and other features enabled by Kotlin.
 
-##Differences from ReactFX
-[ReactFX](https://github.com/TomasMikula/ReactFX) is a popular API to implement reactive patterns with JavaFX using the `EventStream`. However, RxJava uses an `Observable` and the two are not (directly) compatible with each other. 
-
-Although ReactFX has some asynchronous operators like `threadBridge`, ReactFX emphasizes *synchronous* behavior. This means it encourages keeping events on the JavaFX thread. RxJavaFX, which fully embraces RxJava and *asynchronous* design, can switch between threads and schedulers with ease.  As long as subscriptions affecting the UI are observed on the JavaFX thread, you can leverage the powerful operators and libraries of RxJava safely.
-
-If you are heavily dependent on RxJava, asynchronous processing, or do not want your entire reactive codebase to be UI-focused, you will probably want to use RxJavaFX. 
 
 ## Comprehensive Example
 ```java

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

@@ -0,0 +1,7 @@
+package rx.javafx.sources;
+
+public enum Flag {
+    ADDED,
+    REMOVED,
+    UPDATED;
+}

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

@@ -0,0 +1,28 @@
+package rx.javafx.sources;
+
+/**
+ * Holds an ADDED, REMOVED, or UPDATED flag with the associated value
+ * @param <T>
+ */
+public final class ListChange<T> {
+    private final T value;
+    private final Flag flag;
+
+    private ListChange(T value, Flag flag) {
+        this.value = value;
+        this.flag = flag;
+    }
+    public static <T> ListChange<T> of(T value, Flag flag) {
+        return new ListChange<>(value, flag);
+    }
+    public T getValue() {
+        return value;
+    }
+    public Flag getFlag() {
+        return flag;
+    }
+    @Override
+    public String toString() {
+        return flag + " " + value;
+    }
+}

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

@@ -191,39 +191,17 @@ public int remove(T value) {
             Integer prev = counts.get(value);
             if (prev != null && prev > 0) {
                 int newVal = prev - 1;
-                counts.put(value, newVal);
+                if (newVal == 0) {
+                    counts.remove(value);
+                } else {
+                    counts.put(value, newVal);
+                }
                 return newVal;
             }
             else {
                 throw new IllegalStateException();
             }
         }
     }
-    public static final class ListChange<T> {
-        private final T value;
-        private final Flag flag;
 
-        private ListChange(T value, Flag flag) {
-            this.value = value;
-            this.flag = flag;
-        }
-        public static <T> ListChange<T> of(T value, Flag flag) {
-            return new ListChange<>(value, flag);
-        }
-        public T getValue() {
-            return value;
-        }
-        public Flag getFlag() {
-            return flag;
-        }
-        @Override
-        public String toString() {
-            return flag + " " + value;
-        }
-    }
-    public enum Flag {
-        ADDED,
-        REMOVED,
-        UPDATED;
-    }
 }

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

@@ -17,6 +17,7 @@
 
 
 import javafx.beans.value.ObservableValue;
+import javafx.collections.ObservableList;
 import javafx.event.ActionEvent;
 import javafx.event.Event;
 import javafx.event.EventType;
@@ -27,6 +28,7 @@
 import javafx.stage.Window;
 import javafx.stage.WindowEvent;
 import rx.Observable;
+import rx.functions.Func1;
 import rx.javafx.sources.*;
 
 
@@ -117,4 +119,73 @@ public static Observable<ActionEvent> fromActionEvents(final ContextMenu context
     public static Observable<ActionEvent> fromActionEvents(final MenuItem menuItem) {
         return ActionEventSource.fromActionEvents(menuItem);
     }
+
+    /**
+     * Creates an observable that emits an ObservableList every time it is modified
+     *
+     * @param source      The target ObservableList of the ListChange events
+     * @return An Observable emitting the ObservableList each time it changes
+     */
+    public static <T> Observable<ObservableList<T>> fromObservableList(final ObservableList<T> source) {
+        return ObservableListSource.fromObservableList(source);
+    }
+    /**
+     * Creates an observable that emits all additions to an ObservableList
+     *
+     * @param source      The target ObservableList for the item add events
+     * @return An Observable emitting items added to the ObservableList
+     */
+    public static <T> Observable<T> fromObservableListAdds(final ObservableList<T> source) {
+        return ObservableListSource.fromObservableListAdds(source);
+    }
+    /**
+     * Creates an observable that emits all removal items from an ObservableList
+     *
+     * @param source      The target ObservableList for the item removal events
+     * @return An Observable emitting items removed from the ObservableList
+     */
+    public static <T> Observable<T> fromObservableListRemovals(final ObservableList<T> source) {
+        return ObservableListSource.fromObservableListRemovals(source);
+    }
+    /**
+     * Creates an observable that emits all updated items from an ObservableList.
+     * If you declare an ObservableList that listens to one or more properties of each element,
+     * you can emit the changed items every time these properties are modified
+     * <pre>ObservableList<Person> sourceList = FXCollections.observableArrayList(user -> new javafx.beans.Observable[]{user.age} );</pre>
+     * @param source      The target ObservableList for the item update events
+     *
+     * @return An Observable emitting items updated in the ObservableList
+     */
+    public static <T> Observable<T> fromObservableListUpdates(final ObservableList<T> source) {
+        return ObservableListSource.fromObservableListUpdates(source);
+    }
+
+    /**
+     * Emits all added, removed, and updated items from an ObservableList
+     * @param source
+     * @return An Observable emitting changed items with an ADDED, REMOVED, or UPDATED flags
+     */
+    public static <T> Observable<ListChange<T>> fromObservableListChanges(final ObservableList<T> source) {
+        return ObservableListSource.fromObservableListChanges(source);
+    }
+    /**
+     * Emits distinctly  added and removed items from an ObservableList.
+     * If dupe items with identical hashcode/equals evaluations are added to an ObservableList, only the first one will fire an ADDED item.
+     * When the last dupe is removed, only then will it fire a REMOVED item.
+     * @param source
+     * @return An Observable emitting changed items with an ADDED, REMOVED, or UPDATED flags
+     */
+    public static <T> Observable<ListChange<T>> fromObservableListDistinctChanges(final ObservableList<T> source) {
+        return ObservableListSource.fromObservableListDistinctChanges(source);
+    }
+    /**
+     * Emits distinctly  added and removed mappings of each item from an ObservableList.
+     * If dupe mapped items with identical hashcode/equals evaluations are added to an ObservableList, only the first one will fire an ADDED item.
+     * When the last dupe is removed, only then will it fire a REMOVED item.
+     * @param source
+     * @return An Observable emitting changed mapped items with an ADDED, REMOVED, or UPDATED flags
+     */
+    public static <T,R> Observable<ListChange<R>> fromObservableListDistinctMappings(final ObservableList<T> source, Func1<T,R> mapper) {
+        return ObservableListSource.fromObservableListDistinctMappings(source,mapper);
+    }
 }

src/main/java/rx/schedulers/JavaFxScheduler.java

@@ -1,18 +1,3 @@
-/**
- * Copyright 2014 Netflix, Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
 package rx.schedulers;
 
 import javafx.animation.KeyFrame;
@@ -163,10 +148,9 @@ public Subscription schedule(final Action0 action) {
                 if (trampoline(runnable)) {
                     return Subscriptions.unsubscribed();
                 }
-                else {
-                    queue.offer(runnable);
-                    Platform.runLater(this);
-                }
+            }else {
+                queue.offer(runnable);
+                Platform.runLater(this);
             }
 
             // wrap for returning so it also removes it from the 'innerSubscription'