Over a million developers have joined DZone.
{{announcement.body}}
{{announcement.title}}

Handling RxJava Observables in a Web UI

DZone's Guide to

Handling RxJava Observables in a Web UI

Learn how to integrate an RxJava Observable with a ZK application while limiting the UI update frequency during a hot event stream.

· Java Zone
Free Resource

Learn how to troubleshoot and diagnose some of the most common performance issues in Java today. Brought to you in partnership with AppDynamics.

Before we dive into the meat of this article, let's start with some definitions.

RxJava: A Java VM implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences. It is popular, as it addresses the concerns about low-level threading, synchronization, thread-safety and concurrent data structures.

ZK Framework: An open source Java Web framework. It is a server-centric user interface solution that enables developers to create highly interactive enterprise applications. I am using this UI framework here because its MVVM pattern and WebSocket support fit well into RxJava's reactive streams.

The Mission

Image titleMoody Robots

Assume you are working on a project on a Robot Farm (I think there could be worse projects). Unfortunately, those Robots are a bit moody and unreliable sometimes — so, of course, there needs to be a supervisor sitting in his/her office chair and watching a screen to monitor all the Robot movements. Since human supervisors are prone to errors too, there should be multiple supervisors observing the same Robots simultaneously.

Based on their current assignment and in order to preserve bandwidth, supervisors must be able to track certain Robots in near real-time (filtered by mood or position) without completely losing track of the overall situation, i.e. The Robots we are interested in have a faster update rate than others (e.g. 100 ms vs. 1 sec).

However, the Robots are constantly sending data at a high frequency to your back-end process. Your challenge is to connect the UI to the stream of information, reducing it based on the filter criteria and throttling the sheer number of events to something the human eye and your network connection can handle.

The Outset

For the Robot event stream, we choose RxJava because it implements a flexible Observable API — ideal when dealing with asynchronous event streams in combination with a powerful set of operators to transform the results. For the Web UI, we go with the open source ZK Framework. RxJava's reactive streams fit well into ZK's MVVM design pattern, making an interesting combination worth talking about.

Let's go over the technologies used.

The Backend (RX Observable)

The stream of  TrackEvent<Robot>-objects is produced by a single hot Observable initialized at startup.

The Observable constantly emits TrackEvents at 10ms intervals (unreliability simulated by randomly updating ~30% of the robots with an additional 2% chance to change the mood) and allows for multiple consumers (using observable.publish() / observable.connect() ).

As this Observable is unaware of the front end, it will just keep emitting the events once started no matter what. For each new subscriber, it will initially send TrackEvents for all Robots followed by the random stream of events every subscriber shares.

 zk.rx.demo.service.RobotBackend:

public class RobotBackend {
    private static int NUM_ROBOTS = 20;
    ...

    /**
     * called once to start the stream of events
     */
    public void start() {
        allRobots = LongStream.range(0L, NUM_ROBOTS)
                .mapToObj(index -> new Robot(100 + index, new Position(0, 0), Robot.Mood.NEUTRAL))
                .collect(Collectors.toConcurrentMap(Robot::getId, robot -> robot));

        Observable<TrackEvent<Robot>> obs = Observable.create(this::backgroundThread);
        hotRobotObservable = obs.publish();
        disposable = hotRobotObservable.connect();
    }

    /**
     * called by each subscriber to connect to the same event stream
     * @return Observable of {@link TrackEvent}
     */
    public Observable<TrackEvent<Robot>> trackRobots() {
        Stream<TrackEvent<Robot>> currentRobots = allRobots.values().stream()
                .map(robot -> new TrackEvent<>(TrackEvent.Name.ON_ENTER, robot, robot));
        //prepend initial state for all robots to the hot stream of updates
        return hotRobotObservable
                .startWith(currentRobots::iterator);
    }

    ... some logic to create the thread and random positions below ...


Just imagine the GPS beacons in our Robots constantly keep sending positional information, which we don't intend to control from the front end side. We simply need to deal with any given update frequency.

The UI (ZK MVVM application)

A simple UI is implemented in ZK using a zul template and a Java ViewModel class. UI-specific calculated properties such as styleClasses (derived from mood and realTime status) are added by wrapping the domain class Robot into a UiRobot.

index.zul:

<?style src="style.css"?>
<zk xmlns:w="client">
    <div id="robotFarm" viewModel="@id('vm') @init('zk.rx.demo.vm.RobotFarmViewModel')">
        ...
        <div sclass="trackingArea">
            <if test="@load(vm.centerRegionTracking)">
                <div sclass="centerRegionArea"/>
            </if>
            <forEach items="@init(vm.trackedRobots)" var="mapEntry">
                <apply uiRobot="@init(mapEntry.value)">
                    <div sclass="@load(uiRobot.styleClasses)"
                         left="@load((uiRobot.robot.position.x += '%'))"
                         top="@load((uiRobot.robot.position.y += '%'))">
                    </div>
                </apply>
            </forEach>
        </div>

        <div sclass="controlArea" align="center">
            Real-time:
            <combobox readonly="true" model="@init(vm.filterNamesModel)" 
                      onSelect="@command('selectFilter')" width="120px"/>
            <button iconSclass="@load(vm.running ? 'z-icon-stop' : 'z-icon-play')" 
                    label="@load(vm.running ? 'Stop' : 'Start')"
                    onClick="@command('toggleRunning')"/>
            <button iconSclass="z-icon-retweet" label="Ping Server" 
                    onClick="@command('testServerResponse')"/>
        </div>
    </div>
</zk>


Lines 12-15: Render Robots as divs with dynamic styles and positions reacting to model changes.

I'll not go too deep into ZK specifics here. Updating the UI (i.e. responding to data changes in the View Model) can be triggered several ways: A simple one is annotating a command handler method with @NotifyChange ...

zk.rx.demo.vm.RobotFarmViewModel:

@Command
@NotifyChange("centerRegionTracking")
public void selectFilter() {
    currentFilter = availableFilters.get(filterNamesModel.getSelection().iterator().next());
    if(isRunning()) {
        start();
    }
}


... which will then update the corresponding data binding @load(vm.centerRegionTracking) in the zul file above:

<if test="@load(vm.centerRegionTracking)">
    <div sclass="centerRegionArea"/>
</if>


An alternative is to call BindUtils.postNotifyChange(String, String, Object, String) directly in order to trigger a @load binding (e.g. called from the ViewModel when a robot is updated).

For the interested, here is the complete MVVM documentation on notifications.

The Magic in the Middle

Knowing how to update the UI is mostly a technicality and will vary between frameworks. The trickier decisions are: when and what to render in the UI because these will have a direct impact on the performance and responsiveness of your application.

Also, since the server and client side are connected via network, there's a latency that needs to be considered. In this example, we use a WebSocket to minimize that overhead. By enabling this feature, the framework will handle this transparently and we don't need to worry about it anymore.

The How: Updating the UI

As the observable emits  TrackEvent<Robot> objects. The basic way to process those might look like this:

backend.trackRobots()
    .subscribe(event -> updateUi(event), this::handleError);


However, this would not work just like that: A technical requirement is to obtain a "lock" before any UI elements in a page (called "desktop" in ZK) can be updated, e.g. via change notification — mentioned above. For user-triggered events, such as mouse or keyboard events, this happens automatically — background threads have to obtain a lock on demand. Especially if only parts of the background thread need to update the UI, the remaining code can run in parallel without blocking user interactions. Getting the "lock" (also called "activating" the desktop) looks as simple as that (similar to transaction boundaries):

try {
  Executions.activate(desktop); //obtains the lock
  //do any UI updates between activate/deactivate <---------------------------
} finally {
  Executions.deactivate(desktop); //will release the lock 
  //and flush the changes to the UI and out to the browser
}


However, this looks like tedious boilerplate code and forgetting to "deactivate" may lead to infinite deadlocks for that particular "desktop". Better we wrap that in some way so it can be reused safely and easily integrated into the observable chain.

Obviously, activate/deactivate don't affect the data of the stream so the RX side effect operators (doOn...) sound like a good match:

backend.trackRobots()
    .observeOn(Schedulers.io())
    .doOnNext(event -> Executions.activate(desktop)) //potentially blocking that's why Schedulers.io()
    .doAfterNext(event -> Executions.deactivate(desktop))
    .doOnTerminate(() -> Executions.deactivate(desktop));
    .subscribe(event -> trackRobot(event), this::handleError);


(Also covered in "RxJava's Side Effect Methods").

Again: Adding those 4 lines before the update might be tedious and still error prone (e.g. using the wrong Scheduler might lead to deadlocks as well as forgetting any of the lines). To make this more manageable, we can reuse those 4 lines by implementing a single activated()-operator extending from ObservableTransformer<T, T>.

public class ZkObservable {
    public static <T> ObservableTransformer<T, T> activated() {
        return activated(Executions.getCurrent().getDesktop());
    }

    public static <T> ObservableTransformer<T, T> activated(Desktop desktop) {
        ZkDesktopOps desktopOps = new ZkDesktopOps(desktop);
        //use IO Scheduler - potentially blocking operation to wait for desktop activation
        return upstream -> upstream
                .observeOn(Schedulers.io())
                .doOnNext(toConsumer(desktopOps.activate())) //potentially blocking
                .doAfterNext(toConsumer(desktopOps.deactivate()))
                .doOnTerminate(desktopOps.deactivate());
    }


Adding this into our chain via Observable.compose() ...

backend.trackRobots()
    .compose(ZkObservable.activated()) //means perform the downstream with an activated desktop
    .subscribe(event -> trackRobot(event), this::handleError);


... the code looks much better, and as a bonus, the same technique can be used for other side effects such as logging, setting thread locals, opening/closing other resources, transaction bracketing etc. — basically anything that doesn't affect the data in the stream.

The When: Throttling/Batching

Being just a side effect doesn't mean it's a cheap side effect. Opening/closing a transaction or acquiring a lock is often expensive and may require waiting (blocking). In our case, it even triggers network communication before the next thread can activate the Desktop again.

This penalty is multiplied by the high update rate from the back-end: ~600 events / sec (20 Robots * 100 updates/second * 30%).
As network round trips tend to take longer than 1.6ms (1000ms/600), we need to reduce the update frequency by processing multiple updates within the same activation/deactivation window. Here the RX Buffer operator does the trick:

backend.trackRobots()
    .buffer(100, TimeUnit.MILLISECONDS))
    .compose(ZkObservable.activated()) //only activate once for all events collected within 100ms
    .subscribe(/*now we'd have a List<TrackEvent<Robot>> here*/)


Now things are already a lot better: Collecting events for 100ms and updating all affected robots at once will save a majority of desktop activations: 10/sec vs ~600/sec before.

Given the chance of an empty buffer, it makes sense to avoid the desktop activation sometimes.

Finally, it would feel more natural to deal with single events (instead of lists) at the end of our stream, which can be done with the flatMap/concatMap operator family.So an improved version would look like this ...

backend.trackRobots()
    .buffer(100, TimeUnit.MILLISECONDS))
    .filter(items -> !items.isEmpty()) //avoid activation when buffer is empty
    .compose(ZkObservable.activated())
    .concatMapIterable(items -> items); //concatMap... preserves the original emission order (flatMap... does not)
    .subscribe(event -> trackRobot(event), this::handleError);


... which can again be composed into the activatedThrottle-operator (combining the buffering, activation and buffer-separation) resulting in this ...

backend.trackRobots()
    .compose(ZkObservable.activatedThrottle(100))
    .subscribe(event -> trackRobot(event), this::handleError);


Now we start talking ... but wait ... there's still more eliminate.

Optimizing the Buffer Size

Batching an average of 60 updates for 20 Robots must inevitably contain duplicates. It would be a waste to update the same Robot several times within a batch, causing redundant render instructions for the client engine and increasing the network traffic and the client side render time.

We are only interested in the latest updates, so we better eliminate duplicates in each batch before rendering. Ideally, this should happen while buffering the events.

So far, I haven't found a default operator that does what I needed (e.g. distinct does quite the opposite, emitting only the first unique event — whereas I'd like to keep the last of a kind — which, of course, doesn't make sense in an infinite stream) so I came up with the following solution using another flavor of the buffer operator accepting a custom buffer supplier (better ideas are welcome!).

The buffer supplier provides a KeyedSet that identifies events for the same Robot via keySelector and preserves the order based on last access. Adding this to a (let's call it) activatedThrottleUnique operator, we get this:

backend.trackRobots()
    //use the robot.id as unique key
    .compose(ZkObservable.activatedThrottleUnique(100, event -> event.getCurrent().getId()))
    .subscribe(event -> trackRobot(event), this::handleError);


Finally, the maximum batch size is limited to 20 (1 update per Robot, no duplicates). Considering where we came from, this dramatically reduces the overall workload, network traffic, and locking overhead. Even if the back-end suddenly produces more events, this number of 20 items per batch would not increase, keeping the UI update effort stable.

Now let's do something for the user.

The What: Filtering

Still, there's only so much a connection and the human eye can handle, which makes it reasonable to think about filtering the event stream and only display what the Robot supervisor is interested in: i.e. highlight only happy/angry robots or follow robots that cross the center region.

Image title

We decide to render the highlighted robots near real-time (updated every 100ms) and render the remaining robots slightly transparent, a bit smaller, and update them only once per second. The highlighting is done by CSS classes, which are added/removed dynamically.
The different update speeds can be achieved by subscribing to the stream twice with different throttle intervals and filters. A CompositeDisposable simplifies subscription cancellation.

Here is the final code as in the example RobotFarmViewModel:

subscriptions = new CompositeDisposable();

Disposable realtimeSubscription = robotTracker.trackRobots(this.currentFilter)
        .compose(ZkObservable.activatedThrottleUnique(100, event -> event.getCurrent().getId()))
        .subscribe(this::trackRealtimeRobot, this::handleError);

Disposable delayedSubscription = robotTracker.trackRobots(negate(this.currentFilter))
        .compose(ZkObservable.activatedThrottleUnique(1000, event -> event.getCurrent().getId()))
        .subscribe(this::trackDelayedRobot, this::handleError);

subscriptions.addAll(realtimeSubscription, delayedSubscription);


  • Lines 3, 7: The RobotTracker class translates the TrackEvents based on a filter

  • Lines 5, 9: The logic to update the UI is slightly different between the real-time and delayed Robots

The filtering is handled by the RobotTracker class, which translates the stream of ON_UPDATE events into ON_UPDATEON_ENTER, and ON_LEAVE to indicate whether the filter condition (e.g. the center region) was entered or left or remained unchanged.
Image title

Now only the currently filtered robots are moving near real time (every 100ms) while the remaining robots only update once every second.

Summary

I hope the Robot Farm supervisors will have as much fun using the streamlined UI as I had implementing it. And of course, I hope to contribute some useful ideas and techniques when using reactive streams glued together with a UI framework, be it ZK or a different one.
As always, some bugs/problems might have slipped into the code, so I am eager to hear your comments and suggestions to improve this experiment.

You might also ask: Why Robots'??'

The answer is simple: Most examples on RX I've seen focus on stock price tickers, which are quite 1-Dimensional. So I thought it might not hurt to add multiple criteria, which both change and can be used to filter on.

Finally, I just love the colorful result compared to plain numbers updating.

Example Sources

The code examples are available on GitHub in the zk-rxdemo repository.

Running the Example

Clone the repo:

 git clone git@github.com:zkoss-demo/zk-rxdemo.git 

The example WAR file can be built using the gradle-wrapper (on Windows simply omit the prefix './'):

 ./gradlew war 

Execute using jetty-runner (fastest):

 ./gradlew startJettyRunner 

Execute using gretty:

 ./gradlew appRun 

Then access the example: http://localhost:8080/zk-rxdemo

Understand the needs and benefits around implementing the right monitoring solution for a growing containerized market. Brought to you in partnership with AppDynamics.

Topics:
reactive extensions ,zk ,observable ,reactive streams ,java ,rxjava ,tutorial

Opinions expressed by DZone contributors are their own.

{{ parent.title || parent.header.title}}

{{ parent.tldr }}

{{ parent.urlSource.name }}