Merge pull request #55 from protogenes/feature/sentinel-null

allow a custom null-sentinel value

Author: thomasnield

src/main/java/io/reactivex/rxjavafx/observables/JavaFxObservable.java

@@ -65,6 +65,18 @@ public static <T> Observable<T> valuesOf(final ObservableValue<T> fxObservable)
         return ObservableValueSource.fromObservableValue(fxObservable);
     }
 
+    /**
+     * Create an rx Observable from a JavaFX ObservableValue
+     *
+     * @param fxObservable the observed ObservableValue
+     * @param nullSentinel the default sentinel value emitted when the observable is null
+     * @param <T>          the type of the observed value
+     * @return an Observable emitting values as the wrapped ObservableValue changes, null will be replaces with nullSentinel
+     */
+    public static <T> Observable<T> valuesOf(final ObservableValue<T> fxObservable, final T nullSentinel) {
+        return ObservableValueSource.fromObservableValue(fxObservable, nullSentinel);
+    }
+
     /**
      * Create an rx Observable from a JavaFX ObservableValue
      *

src/main/java/io/reactivex/rxjavafx/sources/ObservableValueSource.java

@@ -42,6 +42,29 @@ public static <T> Observable<T> fromObservableValue(final ObservableValue<T> fxO
         });
     }
 
+    public static <T> Observable<T> fromObservableValue(final ObservableValue<T> fxObservable, final T nullSentinel) {
+        if (nullSentinel == null) {
+            throw new NullPointerException("The null value sentinel must not be null.");
+        }
+        return Observable.create((ObservableEmitter<T> emitter) -> {
+            if (fxObservable.getValue() != null) {
+                emitter.onNext(fxObservable.getValue());
+            }
+
+            final ChangeListener<T> listener = (observableValue, prev, current) -> {
+                if (current != null) {
+                    emitter.onNext(current);
+                } else {
+                    emitter.onNext(nullSentinel);
+                }
+            };
+
+            fxObservable.addListener(listener);
+
+            emitter.setDisposable(JavaFxSubscriptions.unsubscribeInEventDispatchThread(() -> fxObservable.removeListener(listener)));
+        });
+    }
+
     public static <T> Observable<Optional<T>> fromNullableObservableValue(final ObservableValue<T> fxObservable) {
         return Observable.create((ObservableEmitter<Optional<T>> emitter) -> {
             emitter.onNext(Optional.ofNullable(fxObservable.getValue()));