Start implementation of BindingSubscriber

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

@@ -0,0 +1,120 @@
+/**
+ * Copyright 2016 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.subscribers;
+
+import com.sun.javafx.binding.ExpressionHelper;
+import javafx.beans.InvalidationListener;
+import javafx.beans.binding.Binding;
+import javafx.beans.value.ChangeListener;
+import javafx.beans.value.ObservableValue;
+import javafx.collections.ObservableList;
+import rx.Subscriber;
+import rx.Subscription;
+import rx.functions.Action1;
+
+
+final class BindingSubscriber<T> extends Subscriber<T> implements ObservableValue<T>, Binding<T>, Subscription {
+
+    private final Action1<Throwable> onError;
+    private ExpressionHelper<T> helper;
+    private T value;
+
+    BindingSubscriber(final Action1<Throwable> onError) {
+        this.onError = onError;
+    }
+    @Override
+    public void onCompleted() {
+        //do nothing
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        onError.call(e);
+    }
+
+    @Override
+    public void onNext(T t) {
+        value = t;
+        fireValueChangedEvent();
+    }
+    @Override
+    public T getValue() {
+        return value;
+    }
+    @Override
+    public boolean isValid() {
+        return true;
+    }
+
+    @Override
+    public void invalidate() {
+        //does nothing
+    }
+
+    @Override
+    public ObservableList<?> getDependencies() {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void dispose() {
+        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

@@ -0,0 +1,36 @@
+/**
+ * Copyright 2016 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.subscribers;
+
+import javafx.beans.binding.Binding;
+import rx.Observable;
+import rx.functions.Action1;
+
+public enum JavaFxSubscriber {
+    ;//no instances
+
+    public static <T> Binding<T> toBinding(Observable<T> obs) {
+        BindingSubscriber<T> bindingSubscriber = new BindingSubscriber<>(e -> {});
+        obs.subscribe(bindingSubscriber);
+        return bindingSubscriber;
+    }
+    public static <T> Binding<T> toBinding(Observable<T> obs, Action1<Throwable> onErrorAction ) {
+        BindingSubscriber<T> bindingSubscriber = new BindingSubscriber<>(onErrorAction);
+        obs.subscribe(bindingSubscriber);
+        return bindingSubscriber;
+    }
+}