Major changes to implementation of BindingSubscriber, but how to dispose binding?

Author: thomasnield

src/main/java/rx/subscribers/BindingSubscriber.java

@@ -16,44 +16,40 @@
 
 package rx.subscribers;
 
+import com.sun.javafx.binding.ExpressionHelper;
+import javafx.beans.InvalidationListener;
 import javafx.beans.binding.Binding;
-import javafx.beans.value.ObservableValueBase;
+import javafx.beans.value.ChangeListener;
+import javafx.beans.value.ObservableValue;
 import javafx.collections.ObservableList;
-import rx.Observable;
 import rx.Subscriber;
+import rx.Subscription;
 import rx.functions.Action1;
 
-import java.util.Optional;
 
+public final class BindingSubscriber<T> extends Subscriber<T> implements ObservableValue<T>, Binding<T>, Subscription {
 
-public final class BindingSubscriber<T> extends ObservableValueBase<T> implements Binding<T> {
-
-    private final Subscriber<T> subscriber;
-
+    private final Action1<Throwable> onError;
+    private ExpressionHelper<T> helper;
     private T value;
 
-    BindingSubscriber(Observable<T> observable, Optional<T> initialValue, final Action1<Throwable> onError) {
-
-        this.subscriber = new Subscriber<T>() {
-            @Override
-            public void onCompleted() {
-                //do nothing
-            }
-
-            @Override
-            public void onError(Throwable e) {
-                onError.call(e);
-            }
-
-            @Override
-            public void onNext(T item) {
-                value = item;
-                fireValueChangedEvent();
-            }
-        };
+    BindingSubscriber(final Action1<Throwable> onError) {
+        this.onError = onError;
+    }
+    @Override
+    public void onCompleted() {
+        //do nothing
+    }
 
-        observable.subscribe(subscriber);
+    @Override
+    public void onError(Throwable e) {
+        onError.call(e);
+    }
 
+    @Override
+    public void onNext(T t) {
+        value = t;
+        fireValueChangedEvent();
     }
     @Override
     public T getValue() {
@@ -76,6 +72,49 @@ public ObservableList<?> getDependencies() {
 
     @Override
     public void dispose() {
-        subscriber.unsubscribe();
+        this.unsubscribe();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void addListener(InvalidationListener listener) {
+        helper = ExpressionHelper.addListener(helper, this, listener);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void addListener(ChangeListener<? super T> listener) {
+        helper = ExpressionHelper.addListener(helper, this, listener);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void removeListener(InvalidationListener listener) {
+        helper = ExpressionHelper.removeListener(helper, listener);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void removeListener(ChangeListener<? super T> listener) {
+        helper = ExpressionHelper.removeListener(helper, listener);
+    }
+
+    /**
+     * Notify the currently registered observers of a value change.
+     *
+     * This implementation will ignore all adds and removes of observers that
+     * are done while a notification is processed. The changes take effect in
+     * the following call to fireValueChangedEvent.
+     */
+    protected void fireValueChangedEvent() {
+        ExpressionHelper.fireValueChangedEvent(helper);
     }
 }

src/main/java/rx/subscribers/JavaFxSubscriber.java

@@ -16,24 +16,20 @@
 
 package rx.subscribers;
 
-import rx.Observable;
+import javafx.beans.property.Property;
 import rx.functions.Action1;
 
-import java.util.Optional;
-
 public enum JavaFxSubscriber {
     ;//no instances
 
-    public static <T> BindingSubscriber<T> toBinding(Observable<T> observable) {
-        return new BindingSubscriber<T>(observable, Optional.empty(), e -> {});
-    }
-    public static <T> BindingSubscriber<T> toBinding(Observable<T> observable,Action1<Throwable> onErrorAction ) {
-        return new BindingSubscriber<T>(observable, Optional.empty(), onErrorAction);
-    }
-    public static <T> BindingSubscriber<T> toBinding(Observable<T> observable, T initialValue, Action1<Throwable> onErrorAction ) {
-        return new BindingSubscriber<T>(observable, Optional.of(initialValue), onErrorAction);
+    public static <T> BindingSubscriber<T> toBinding(Property<? super T> t) {
+        BindingSubscriber<T> bindingSubscriber = new BindingSubscriber<>(e -> {});
+        t.bind(bindingSubscriber);
+        return bindingSubscriber;
     }
-    public static <T> BindingSubscriber<T> toBinding(Observable<T> observable, T initialValue) {
-        return new BindingSubscriber<T>(observable, Optional.of(initialValue), e -> {});
+    public static <T> BindingSubscriber<T> toBinding(Property<? super T> t, Action1<Throwable> onErrorAction ) {
+        BindingSubscriber<T> bindingSubscriber = new BindingSubscriber<>(onErrorAction);
+        t.bind(bindingSubscriber);
+        return bindingSubscriber;
     }
 }