Lazy BindingSubscriber

[thomasnield]

There should be an option to make the RxJavaFX Binding implementation lazy. That way it delays subscribing to the Observable until the Binding is first called for the value.

Maybe there needs to be two implementations: EagerBindingSubscriber and LazyBindingSubscriber. The first one is what exists currently as BindingSubscriber.

[thomasnield]

Do you have any thoughts on this @raniejade?

[raniejade]

I don't see any use case that needs a lazy binding, normally when you change an Observable into a binding you would use it immediately. Nonetheless, let's keep this open as others might have a good use case. Only cold and highly composed Observables would benefit from this, performance wise.

[thomasnield]

I'll need to validate this as I might be assuming... I was thinking of TableView and other data controls benefiting from this. If each TableView cell uses a Binding to Subscribe to a cold, expensive Observable, maybe there is benefit if subscriptions only happen to cells currently in view.

This may not be the case though and if a TableView calls the ObservableValues eagerly for all records, and not just the ones in view, that use case is void.

[thomasnield]

Hey @edvin, I know you are modest about your RxJava knowledge but you've been helpful many times. If a Binding lazily subscribes to an Observable when it is first needed (thus delaying expensive operations) , do you see use cases for that?

@hastebrot, @t-boom I'd value your guys' thoughts too if you have any.

If I recall, JavaFX likes to do Bindings lazily... that's why there is a invalidation communication that signals to re-get the value when it's next called. It seems apt to extend this behavior for RxJava too, which also is lazy.

[thomasnield]

Never mind I was right. I created a lazy implementation just to test it out with a TableView.

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.Observable;
import rx.Subscription;


final class LazyBindingSubscriber<T> implements ObservableValue<T>, Binding<T> {

    private final Observable<T> observable;
    private final BindingSubscriber<T> binding;
    private Subscription subscription;

    LazyBindingSubscriber(Observable<T> observable, BindingSubscriber<T> binding) {
        this.observable = observable;
        this.binding = binding;
    }
    @Override
    public T getValue() {
        if (subscription == null) {
            subscription = observable.subscribe(binding);
        }
        return binding.getValue();
    }
    @Override
    public boolean isValid() {
        return binding.isValid();
    }

    @Override
    public void invalidate() {
        binding.invalidate();
    }

    @Override
    public ObservableList<?> getDependencies() {
        throw new UnsupportedOperationException();
    }

    @Override
    public void dispose() {
        if (subscription != null) {
            binding.dispose();
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void addListener(InvalidationListener listener) {
        binding.addListener(listener);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void addListener(ChangeListener<? super T> listener) {
        binding.addListener(listener);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void removeListener(InvalidationListener listener) {
        binding.removeListener(listener);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void removeListener(ChangeListener<? super T> listener) {
        binding.removeListener(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() {
        binding.fireValueChangedEvent();
    }
}

And sure enough, when I used it in a TableView, I got print lines for only the visible records, and I got more when I scrolled. This proves the TableView is lazy, and might benefit from a lazy implementation of BindingSubscriber.

import javafx.application.Application;
import javafx.beans.binding.Binding;
import javafx.scene.Scene;
import javafx.scene.control.TableColumn;
import javafx.scene.control.TableView;
import javafx.scene.layout.GridPane;
import javafx.scene.layout.Priority;
import javafx.stage.Stage;
import rx.Observable;
import rx.subscribers.JavaFxSubscriber;

public final class Test extends Application {

    @Override
    public void start(Stage primaryStage) throws Exception {

        TableView<Item> tableView = new TableView<>();

        TableColumn<Item,Integer> column = new TableColumn<>();
        column.setCellValueFactory(v -> v.getValue().value);
        tableView.getColumns().add(column);

        for (int i = 0; i< 1000; i++){
            tableView.getItems().add(new Item(i));
        }

        GridPane grp = new GridPane();

        GridPane.setFillHeight(tableView, true);
        GridPane.setFillWidth(tableView, true);
        GridPane.setHgrow(tableView, Priority.ALWAYS);
        GridPane.setVgrow(tableView, Priority.ALWAYS);
        grp.getChildren().add(tableView);

        Scene scene = new Scene(grp);

        primaryStage.setScene(scene);

        primaryStage.show();

    }

    private static final class Item {

        final Binding<Integer> value;

        Item(Integer value) {
            this.value =  JavaFxSubscriber.toLazyBinding(Observable.just(value)
                    .doOnNext(v -> System.out.println("Emitting " + v)));
        }

    }

    public static void main(String[] args) {
        Application.launch(args);
    }
}

[thomasnield]

I went ahead and added this to the API. Let me know if you guys have any additional thoughts or ideas, but I think this looks warranted. JavaFX is lazy, RxJava is lazy... it makes sense the interop should be lazy haha. It makes you almost wonder if this should be the default policy actually...

And @raniejade you are right this is not needed for most controls and eager subscription makes sense. But I think data controls are where lazy subscription might become handy.

[edvin]

I think you might actually introduce more overhead with this, at least in the use case above. There might be more warranted use cases though for sure. Use with caution :)

[thomasnield]

You referring to the delegation and double-Binding nesting? I'll see if I can come up with a non-delegated pattern...

[raniejade]

Was going to suggest final class LazyBindingSubscriber<T> extend Subscriber<T> then just modify getValue(), but the value can be null initially. Note that onNext(item) can be called on a different thread.

[thomasnield]

Note that onNext(item) can be called on a different thread.

From the Observable Contract:

Observables must issue notifications to observers serially (not in parallel). They may issue these notifications from different threads, but there must be a formal happens-before relationship between the notifications.

Yeah on anyonNext() it's always been my knee-jerk reaction to make any side-effect variables volatile, until @akarnokd and @davidmoten reminded me that Observables have a strictly serial nature and synchronization is handled by RxJava. This can be undermined if you built bad custom operators or using Subjects. In this case, every update and read to the BindingSubscriber is assumed to be on the JavaFX thread.

But anyway, I just consolidated both Binding implementations into a single one. Hoping this is leaner if I understood @edvin's caution correctly.

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.Observable;
import rx.Subscriber;
import rx.Subscription;
import rx.functions.Action1;


final class BindingSubscriber<T> extends Subscriber<T> implements ObservableValue<T>, Binding<T> {

    private final Observable<T> observable;
    private final Action1<Throwable> onError;
    private Subscription subscription;
    private ExpressionHelper<T> helper;
    private T value;

    private BindingSubscriber(Observable<T> observable, Action1<Throwable> onError) {
        this.observable = observable;
        this.onError = onError;
    }

    public static <T> BindingSubscriber<T> forObservable(Observable<T> observable, Action1<Throwable> onError, boolean isLazy) {
        BindingSubscriber<T> bindingSubscriber = new BindingSubscriber<>(observable, onError);

        if (!isLazy) {
            bindingSubscriber.connect();
        }

        return bindingSubscriber;
    }
    private void connect() {
        subscription = observable.subscribe(this);
    }
    @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() {
        if (subscription == null) {
            connect();
        }
        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() {
        if (subscription != null) {
            subscription.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);
    }
}

[thomasnield]

While we are gutting this, do we see value in adding a parameter for onCompleted and not just onError?

[thomasnield]

Hmmm... interesting. When you use the lazy Binding and scroll quickly through records in the TableView, it will only subscribe to the records you've slowed down on, not the ones you zipped past. This behavior could be desirable to prevent scroll lag.

import javafx.application.Application;
import javafx.beans.binding.Binding;
import javafx.scene.Scene;
import javafx.scene.control.Button;
import javafx.scene.control.TableColumn;
import javafx.scene.control.TableView;
import javafx.scene.layout.BorderPane;
import javafx.stage.Stage;
import javafx.util.Duration;
import rx.Observable;
import rx.observables.JavaFxObservable;
import rx.subscribers.JavaFxSubscriber;

public final class Test extends Application {

    @Override
    public void start(Stage primaryStage) throws Exception {

        TableView<Item> tableView = new TableView<>();

        TableColumn<Item,Long> column = new TableColumn<>();
        column.setCellValueFactory(v -> v.getValue().value);
        tableView.getColumns().add(column);

        for (int i = 0; i< 1000; i++){
            tableView.getItems().add(new Item(i));
        }

        BorderPane pane = new BorderPane();
        pane.setCenter(tableView);
        Button button = new Button("Stop");
        pane.setLeft(button);

        button.setOnAction(ae -> tableView.getItems().forEach(Item::dispose));

        Scene scene = new Scene(pane);

        primaryStage.setScene(scene);

        primaryStage.show();

    }

    private static final class Item {
        private static final Observable<Long> interval = JavaFxObservable.interval(Duration.seconds(1));
        final Binding<Long> value;

        Item(Integer value) {
            this.value =  JavaFxSubscriber.toLazyBinding(interval
                    .doOnSubscribe(() -> System.out.println("Subscribing " + value))
                    .doOnUnsubscribe(() -> System.out.println("Unsubscribing " + value))
                    , Throwable::printStackTrace);
        }

        public void dispose() {
            value.dispose();
        }

    }

    public static void main(String[] args) {
        Application.launch(args);
    }

}

[edvin]

Hmm.. That sounds neat, but I don't quite understand why it behaves like that. Do you know/can you explain?

[thomasnield]

Sorry, it's not so much my implementation but rather JavaFX itself defining this behavior. The TableView must have been built to be lazy and not request values from records that are scrolled past quickly.

If nobody else has concerns and approves of this feature, I think I'm going to draft up a release by the end of this week. Are we okay with the following signatures being added to the API?

JavaFxObservable.interval(Duration duration)
JavaFxSubscriber.toBinding(Observable<T> observable)
JavaFxSubscriber.toBinding(Observable<T> observable, Action1<Throwable> onError)
JavaFxSubscriber.toLazyBinding(Observable<T> observable)
JavaFxSubscriber.toLazyBinding(Observable<T> observable, Action1<Throwable> onError)

[thomasnield]

Excellent, I'll close and draft up a release.