EventBus/PubSub vs (reactive extensions) RX with respect to code clarity in a single threaded application

Question

Currently, I am using an EventBus/PubSub architecture/pattern with Scala (and JavaFX) to implement a simple note organizing app (sort of like an Evernote client with some added mind mapping functionality) and I have to say that I really like EventBus over the observer pattern.

Here are some EventBus libraries :

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

http://eventbus.org (currently seems to be down) this is the one I am using in my implementation.

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

Here is a comparison of EventBus libraries : http://codeblock.engio.net/37/

EventBus is related to the publish-subscribe pattern.

However !

Recently, I took the Reactive course by Coursera and started to wonder whether using RXJava instead of EventBus would simplify the event handling code even more in a single threaded application ?

I would like to ask about the experiences of people who programmed using both technologies (some kind of eventbus library and some form of the reactive extensions (RX)): was it easier to tackle event handling complexity using RX than with an event bus architecture given that there was no need to use multiple threads ?

I am asking this because I have heard in the Reactive Lectures on Coursera that RX leads to much cleaner code than using the observer pattern (i.e. there is no "callback hell"), however I did not find any comparison between EventBus architecture vs RXJava. So it's clear that both EventBus and RXJava are better than the observer pattern but which is better in a single threaded applications in terms of code clarity and maintainability ?

If I understand correctly the main selling point of RXJava is that it can be used to produce responsive applications if there are blocking operations (e.g. waiting for response from a server).

But I don't care about asychronicity at all, all I care about is keeping the code clean, untangled and easy to reason about in a single threaded application.

In that case, is it still better to use RXJava than EventBus ?

I think EventBus would be a simpler and cleaner solution and I don't see any reason why I should use RXJava for a single threaded application in favour of a simple EventBus architecture.

But I might be wrong!

Please correct me if I am wrong and explain why RXJava would be better than a simple EventBus in case of a single threaded application where no blocking operations are carried out.

Answer 1

I think you have to use the rxjava because it provides much more flexibility. If you need a bus you can use an enum like this:

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 2

The following is what I see as benefits of using reactive event streams in a single-threaded synchronous application.

1. More declarative, less side-effects and less mutable state.

Event streams are capable of encapsulating logic and state, potentially leaving your code without side-effects and mutable variables.

Consider an application that counts button clicks and displays the number of clicks as a label.

Plain Java solution:

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!!!
});

ReactFX solution:

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 mutable variable is used and the side-effectful assignment to label.textProperty() is hidden behind an abstraction.

In his master thesis, Eugen Kiss has proposed integration of ReactFX with Scala. Using his integration, the solution could look like this:

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)

It is equivalent to the previous, with the additional benefit of eliminating inversion of control.

2. Means to eliminate glitches and redundant computations. (ReactFX only)

Glitches are temporary inconsistencies in observable state. ReactFX has means to suspend event propagation until all updates to an object have been processed, avoiding both glitches and redundant updates. In particular, have a look at suspendable event streams, Indicator, InhiBeans and my blog post about InhiBeans. These techniques rely on the fact that event propagation is synchronous, therefore do not translate to rxJava.

3. Clear connection between event producer and event consumer.

Event bus is a global object that anyone can publish to and subscribe to. The coupling between event producer and event consumer is indirect and therefore less clear. With reactive event streams, the coupling between producer and consumer is much more explicit. Compare:

Event bus:

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

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

The relationship between a and consumer is not clear from looking at just the initialization code.

Event streams:

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

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

The relationship between a and consumer is very explicit.

4. Events published by an object are manifested in its API.

Using the example from the previous section, in the event bus sample, A's API does not tell you what events are published by instances of A. On the other hand, in the event streams sample, A's API states that instances of A publish events of type MyEvent.