将Observable转换为ObservableValue/Binding/EventStream的最有效方法

Question

我将更多地使用RxJava和ReactFX，但我试图理解的是如何调和两者，因为ReactFX没有RxJava依赖，所以两者如何在同一个单子内相互交谈?这对于在JavaFX的ObservableValue、RxJava的Observable和ReactFX的StreamEvent之间架设桥梁来说尤其如此，而不需要大量的样板文件。

我想用RxJava组合我的核心业务逻辑，因为它们不会总是支持JavaFX应用程序。但我希望JavaFX UI使用ReactFX和利用EventStream。所以我的问题是，把EventStream变成Observable，把Observable变成EventStream、Binding或ObservableValue，最有效的方法是什么?我知道我可以完全使用RxJava，但我想利用ReactFX平台的线程安全性和便利性…

//DESIRE 1- Turn EventStream into Observable in the same monad
Observable<Foo> obs = EventStream.valuesOf(fooObservableValue).toObservable();
//Desire 2- Turn Observable into ObservableValue, Eventstream, or Binding
Binding<Foo> obsVal = Observable.create(...).toBinding();

Answer 1

将ReactFX EventStream转换为RxJava Observable:

Observable<Foo> toRx(EventStream<Foo> es) {
    PublishSubject<Foo> sub = PublishSubject.create();
    es.subscribe(sub::onNext);
    return sub;
}

将RxJava Observable转换为ReactFX EventStream:

EventStream<Foo> fromRx(Observable<Foo> obs) {
    EventSource<Foo> es = new EventSource<>();
    obs.subscribe(foo -> Platform.runLater(() -> es.push(foo)));
    return es;
}

注意后一种情况下的Platform.runLater(...)。这使得结果EventStream在JavaFX应用程序线程上发出事件。

还要注意，在这两种情况下，我们都忽略了subscribe方法返回的Subscription s。如果要在应用程序的生命周期内建立绑定，那么这样做很好。另一方面，如果它们之间的绑定应该是短暂的，那么在第一种情况下，您将让RxJava组件公开Subject, ReactFX组件公开EventStream，然后根据需要执行subscribe/unsubscribe。第二种情况类似。

Answer 2

我不熟悉ReactFX，但是看看API，我可以推断出这些转换:

public static <T> Observable<T> toObservable(EventStream<? extends T> es) {
    return Observable.create(child -> {
        Subscription s = es.subscribe(child::onNext);
        child.add(Subscriptions.create(s::unsubscribe));
    });
}
public static <T> EventStream<T> toEventStream(Observable<? extends T> o) {
    return new EventStream<T>() {
        final Vector<Consumer<? super T>> observers = new Vector<>();
        @Override
        public void addObserver(Consumer<? super T> observer) {
            observers.add(observer);
        }
        @Override
        public void removeObserver(Consumer<? super T> observer) {
            observers.remove(observer);
        }
        @Override
        public Subscription subscribe(Consumer<? super T> subscriber) {
            addObserver(subscriber);
            rx.Subscriber<T> s = new rx.Subscriber<T>() {
                @Override
                public void onNext(T t) {
                    subscriber.accept(t);
                }
                @Override
                public void onError(Throwable e) {
                    e.printStackTrace();
                    removeObserver(subscriber);
                }
                @Override
                public void onCompleted() {
                    removeObserver(subscriber);
                }
            };
            o.subscribe(s);
            return () -> {
                s.unsubscribe();
                removeObserver(subscriber);
            };
        }
    };
}

两者都应该提供退订功能，尽管ReactFX不支持同步退订，而且我真的不知道EventStream是否可以用作热可观察对象或冷可观察对象。我无法访问Binding，因此无法提供帮助。