event handling - programming - EventBus/PubSub vs(extensiones reactivas) RX con respecto a la claridad del código en una sola aplicación con subprocesos

rxjs introduction

Question

Actualmente, estoy usando una arquitectura / patrón EventBus / PubSub con Scala (y JavaFX) para implementar una aplicación de organización de notas simple (algo así como un cliente de Evernote con alguna funcionalidad de asignación de mente añadida) y tengo que decir que realmente me gusta EventBus el patrón del observador

Aquí hay algunas bibliotecas de EventBus:

https://code.google.com/p/guava-libraries/wiki/EventBusExplained

http://eventbus.org (actualmente parece estar fuera de servicio) este es el que estoy usando en mi implementación.

http://greenrobot.github.io/EventBus/

Aquí hay una comparación de las bibliotecas de EventBus: http://codeblock.engio.net/37/

EventBus está relacionado con el PubSub .

Sin embargo !

Recientemente, tomé el curso Reactive de Coursera y comencé a preguntarme si usar RXJava lugar de EventBus simplificaría aún más el código de manejo de eventos en una sola aplicación de subprocesos .

Me gustaría preguntar sobre las experiencias de personas que programaron usando ambas tecnologías (algún tipo de biblioteca de eventos y alguna forma de extensiones reactivas (RX)): era más fácil abordar la complejidad de manejo de eventos usando RX que con una arquitectura de bus de eventos dada que no había necesidad de usar múltiples hilos ?

Estoy preguntando esto porque escuché en las Lecturas Reactivas en Coursera que RX conduce a un código mucho más limpio que usar el patrón de observador (es decir, no hay un "infierno de devolución de llamada"), sin embargo, no encontré ninguna comparación entre EventBus architecture vs RXJava . Entonces, está claro que tanto EventBus como RXJava son mejores que el patrón de observador, pero ¿ cuál es mejor en una sola aplicación de subprocesos en términos de claridad de código y mantenibilidad?

Si entiendo correctamente, el principal argumento de venta de RXJava es que puede usarse para producir aplicaciones receptivas si hay operaciones de bloqueo (por ejemplo, esperando la respuesta de un servidor).

Pero a mí no me importa la homicidicidad, lo único que me preocupa es mantener el código limpio, desenredado y fácil de razonar en una sola aplicación con subprocesos .

En ese caso, ¿es aún mejor usar RXJava que EventBus?

Creo que EventBus sería una solución más simple y más limpia, y no veo ninguna razón por la que deba usar RXJava para una única aplicación de subprocesos a favor de una arquitectura de EventBus simple.

¡Pero podría estar equivocado!

Corrígeme si me equivoco y explique por qué RXJava sería mejor que un EventBus simple en el caso de una aplicación de un solo subproceso donde no se llevan a cabo operaciones de bloqueo.

Answer 1

Aprendí una o dos cosas desde que hice esta pregunta hace 2 años, esta es mi comprensión actual (como se explica en el book FRP de Stephen):

Ambos intentan ayudar a describir una máquina de estado, es decir, describen cómo cambia el estado del programa en respuesta a los eventos.

La diferencia clave entre EventBus y FRP es la composicionalidad :

• ¿Qué es la composición?

  • La programación funcional es compositiva. Todos podemos estar de acuerdo en eso. Podemos tomar cualquier función pura, combinarla con otras funciones puras y lo que obtenemos es una función pura más complicada.
  • La composicionalidad significa que cuando declaras algo, en el sitio de la declaración se define todo el comportamiento de la entidad declarada .

• FRP es una forma compositiva de describir una máquina de estado, y event-bus no lo es. Por qué ?

  • Describe una máquina de estado.
  • Es composicional porque la descripción se hace usando funciones puras y valores inmutables.

• EventBus no es una forma compositiva de describir una máquina de estado. Por qué no ?

  • No puede tomar dos buses de eventos y componerlos de forma tal que se obtenga un nuevo bus de eventos compuesto que describa la máquina de estados compuesta. Por qué no ?
    • Los buses de eventos no son ciudadanos de primera clase (a diferencia del Evento / Stream de FRP).
      • ¿Qué sucede si intentas hacer que los autobuses de eventos sean ciudadanos de primera clase?
        • Luego obtienes algo parecido a FRP / RX.
    • Estado influenciado por los autobuses de eventos no son
      • ciudadanos de primera clase (es decir, valores referencialmente transparentes, puros en contraste con la conducta / célula de FRP)
      • vinculado a los buses de eventos de una manera declarativa / funcional, en su lugar el estado se modifica de forma imperativa mediante el manejo de eventos

En resumen, EventBus no es compositivo, porque el significado y el comportamiento de un EventBus compuesto (es decir, la evolución temporal del estado que está influenciado por dicho EventBus compuesto) depende del tiempo (es decir, el estado de las partes del software que no están incluidas explícitamente en la declaración del EventBus compuesto). En otras palabras, si intentara declarar un EventBus compuesto, entonces no sería posible determinar (solo mirando la declaración del EventBus compuesto) qué reglas rigen la evolución del estado de aquellos estados que están influenciados por el EventBus compuesto, esto está en contraste con FRP, donde esto se puede hacer).

Answer 2

Creo que debes usar el rxjava porque proporciona mucha más flexibilidad. Si necesita un autobús, puede usar una enumeración como esta:

public enum Events {

  public static PublishSubject <Object> myEvent = PublishSubject.create ();
}

//where you want to publish something
Events.myEvent.onNext(myObject);

//where you want to receive an event
Events.myEvent.subscribe (...);

.

Answer 3

Lo que veo a continuación son los beneficios del uso de secuencias de eventos reactivos en una aplicación síncrona de subproceso único .

1. Más declaraciones, menos efectos secundarios y menos estado mutable.

Las secuencias de eventos son capaces de encapsular la lógica y el estado, dejando potencialmente su código sin efectos secundarios y variables mutables.

Considere una aplicación que cuente los clics del botón y muestre la cantidad de clics como una etiqueta.

Solución JavaFX simple:

private int counter = 0; // mutable field!!!

Button incBtn = new Button("Increment");
Label label = new Label("0");

incBtn.addEventHandler(ACTION, a -> {
    label.setText(Integer.toString(++counter)); // side-effect!!!
});

Solución ReactFX:

Button incBtn = new Button("Increment");
Label label = new Label("0");

EventStreams.eventsOf(incBtn, ACTION)
        .accumulate(0, (n, a) -> n + 1)
        .map(Object::toString)
        .feedTo(label.textProperty());

No se utiliza ninguna variable mutable y la asignación de label.textProperty() a label.textProperty() se oculta detrás de una abstracción.

En su tesis de maestría, Eugen Kiss ha propuesto la integración de ReactFX con Scala. Usando su integración, la solución podría verse así:

val incBtn = new Button("Increment")
val label = new Label("0")

label.text |= EventStreams.eventsOf(incBtn, ACTION)
    .accumulate(0, (n, a) => n + 1)
    .map(n => n.toString)

Es equivalente al anterior, con el beneficio adicional de eliminar la inversión del control.

2. Significa eliminar problemas técnicos y cálculos redundantes. (ReactFX solamente)

Glitches son inconsistencias temporales en estado observable. ReactFX tiene los medios para suspender la propagación de eventos hasta que se hayan procesado todas las actualizaciones de un objeto, evitando fallas y actualizaciones redundantes. En particular, eche un vistazo a las secuencias de eventos suspendibles , Indicator , InhiBeans y mi blog sobre InhiBeans . Estas técnicas se basan en el hecho de que la propagación de eventos es sincrónica, por lo tanto, no se traduce en rxJava.

3. Borrar la conexión entre el productor del evento y el consumidor del evento.

El bus de eventos es un objeto global al que cualquiera puede publicar y suscribirse. El acoplamiento entre el productor del evento y el consumidor del evento es indirecto y, por lo tanto, menos claro. Con flujos de eventos reactivos, el acoplamiento entre el productor y el consumidor es mucho más explícito. Comparar:

Autobús del evento:

class A {
    public void f() {
        eventBus.post(evt);
    }
}

// during initialization
eventBus.register(consumer);
A a = new A();

La relación entre a y el consumer no está clara al mirar solo el código de inicialización.

Secuencias de eventos:

class A {
    public EventStream<MyEvent> events() { /* ... */ }
}

// during initialization
A a = new A();
a.events().subscribe(consumer);

La relación entre a y el consumer es muy explícita.

4. Los eventos publicados por un objeto se manifiestan en su API.

Usando el ejemplo de la sección anterior, en la muestra del bus de eventos, la API de A no le dice qué eventos están publicados por instancias de A Por otro lado, en el ejemplo de secuencias de eventos, la API de A indica que las instancias de A publican eventos de tipo MyEvent .

Answer 4

Según mi comentario anterior, JavaFx tiene una clase ObservableValue que corresponde a RX Observable (probablemente ConnectableObservable para ser más preciso, ya que permite más de una suscripción). Utilizo la siguiente clase implícita para convertir de RX a JFX, así:

import scala.collection.mutable.Map
import javafx.beans.InvalidationListener
import javafx.beans.value.ChangeListener
import javafx.beans.value.ObservableValue
import rx.lang.scala.Observable
import rx.lang.scala.Subscription

/**
 * Wrapper to allow interoperability bewteen RX observables and JavaFX
 * observables. 
 */
object JfxRxImplicitConversion {
  implicit class JfxRxObservable[T](theObs : Observable[T]) extends ObservableValue[T] { jfxRxObs =>
    val invalListeners : Map[InvalidationListener,Subscription] = Map.empty
    val changeListeners : Map[ChangeListener[_ >: T],Subscription] = Map.empty
    var last : T = _
    theObs.subscribe{last = _}

    override def getValue() : T = last 

    override def addListener(arg0 : InvalidationListener) : Unit = {
      invalListeners += arg0 -> theObs.subscribe { next : T => arg0.invalidated(jfxRxObs) }
    }

    override def removeListener(arg0 : InvalidationListener) : Unit = {
      invalListeners(arg0).unsubscribe
      invalListeners - arg0
    }

    override def addListener(arg0 : ChangeListener[_ >: T]) : Unit = {
      changeListeners += arg0 -> theObs.subscribe { next : T => arg0.changed(jfxRxObs,last,next) }
    }

    override def removeListener(arg0 : ChangeListener[_ >: T]) : Unit = {
      changeListeners(arg0).unsubscribe
      changeListeners - arg0
    }
  }
}

Luego, le permite usar enlaces de propiedades como ese (esto es ScalaFX, pero corresponde a Property.bind en JavaFX):

new Label {
    text <== rxObs
}

Donde rxObs podría ser, por ejemplo:

val rxObs : rx.Observable[String] = Observable.
  interval(1 second).
  map{_.toString}.
  observeOn{rx.lang.scala.schedulers.ExecutorScheduler(JavaFXExecutorService)} 

que es simplemente un contador que se incrementa cada segundo. Solo recuerda importar la clase implícita. ¡No puedo imaginar que sea más limpio que eso!

Lo anterior es un poco intrincado, debido a la necesidad de utilizar un planificador que funciona bien con JavaFx. Consulte this pregunta para obtener un enlace a un JavaFXExecutorService de cómo se implementa JavaFXExecutorService . Hay una solicitud de mejora para Scala RX para convertir esto en un argumento implícito, por lo que en el futuro puede que no necesite la llamada .observeOn .