Самый эффективный способ превратить Observable в ObservableValue/Binding/EventStream?

Question

Я буду использовать RxJava и ReactFX более интенсивно, но то, что я пытаюсь понять, состоит в том, как примирить эти два, поскольку у ReactFX нет зависимости RxJava, так как могут обе друг друга разговаривать в одной монаде? Это особенно касается перехода между JavaFX ObservableValue, RxJava и StreamEvent ReactFX без большого количества шаблонов.

Я хочу составить свою основную бизнес-логику с помощью RxJava, поскольку они не всегда будут поддерживать приложения JavaFX. Но я хочу, чтобы интерфейс JavaFX использовал ReactFX и использовал EventStream. Поэтому мой вопрос - это самый эффективный способ превратить EventStream в Observable и Observable в EventStream, Binding, или ObservableValue? Я знаю, что я мог бы просто использовать RxJava через борт, но я хочу использовать потокобезопасность Platform 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.

Также обратите внимание, что мы игнорируем Subscription s, возвращенные из методов subscribe в обоих случаях. Это нормально, если вы устанавливаете привязку на всю жизнь своего приложения. Если, с другой стороны, привязка между ними должна быть недолговечной, в первом случае у вас будет компонент RxJava, выставляющий Subject, ваш компонент ReactFX выставляет EventStream, а затем/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, поэтому не могу вам помочь.