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

我将更加大量地使用RxJava和ReactFX，但是我想了解的是如何调和两者，因为ReactFX没有RxJava依赖关系，那么两者如何在同一个monad中彼此通信？在没有很多样板的情况下，在JavaFX的ObservableValue，RxJava的Observable和ReactFX的StreamEvent之间架桥的情况尤其如此。

我想用RxJava构成我的核心业务逻辑，因为它们并不总是支持JavaFX应用程序。但是我希望JavaFX UI使用ReactFX并利用EventStream。所以我的问题是将EventStream转换为Observable并将Observable转换为EventStream，Binding或ObservableValue的最有效方法是什么？我知道我可以全盘使用RxJava，但我想利用ReactFX的Platform线程安全性和便利性...

//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();

最佳答案

要将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应用程序线程上发出事件。

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