What does Reactive mean?
There are as many definitions of what “reactive” is as there are articles on the subject, but the basic idea behind the concept is that there are entities, called observables or streams, that represent a value that changes over time (or a list of values that might grow endlessly) and there are computations involving these entities that will themselves return a value that changes over time.
That means we can apply transformations or compositions to one or more observables to get a new observable that represents a different value (or list of values).
What is RxJava?
RxJava is a java implementation of the reactive pattern. It was open sourced by Netflix in 2014 and it’s very popular at the moment to manage asynchronous operations in a range of different environments, from Android applications to backend microservices. It’s heavily based on Microsoft Reactive Extensions.
The keystone of RxJava is the Observable data type. It can be thought as an equivalent to Iterable which is push based rather than pull based.
With an Iterable, the consumer pulls values from the producer and the thread is blocked until those values arrive.
Observables, in contrast, pushes values to the consumer whenever values are available. This approach is more flexible, because values can arrive synchronously or asynchronously.
Example
Let’s go through a simple example to introduce the usage of RxJava. We are going to build a web map showing burglaries in Los Angeles in real time. The full code of the finished example can be found at github.
The first step to start building our application will be to parse the CSV file. For that, we will use the apache common library. We want to represent the data into the CSV file as an Observable (stream) of records that can be then transformed into an observable of crimes.
There are multiple ways to create observables. For fixed size sequences Observable.just() can be used:
This constructor is really helpful for testing purposes, or when returning fallback or one-element responses. Similar constructors exist for creating observables that terminate immediately
Or observables that fail immediately.
In our case, the Apache Common library returns an Iterable that we can use to create our Observable calling the Observable.from() method.
The Observable that we just created will emit one element for each record in the CSV file and then it's going to terminate.
The next step on the way to building our awesome map is to extract the information needed from each record and to create an Observable of crimes. In order to do this, we will use the map operator.
The map operator transforms the items emitted by an Observable by applying a function to each item. The result is a new Observable of a different (or the same) type.
Given a method Crime parseCrimeFromCSVRecord(CSVRecord csvRecord) that converts a CSV record into a Crime object we can use the following code to apply the transformation to all the items in the Observable:
(The implementation of the parseCrimeFromCSVRecord won’t be included here because it's trivial and it can be found in the github repository containing the full example code)
At this point we already have an Observable that emits the crimes taken from the CSV, the only problem is that crimes are emitted as fast as they can be read from the CSV and we wanted to process them at a 1/second rate to simulate real time data.
To accomplish this we will first create an Observable that emits a signal once per second using the intervaloperator.
Once we have our signal in place we will use another reactive operator called zip to merge both observables into a new one emitting one crime per second.
The zip operator combines the emissions of multiple Observables together via a specified function and emits single items for each combination based on the results of this function.
In our case the function will simply return the crime object.
It's worth noting that the rate at which both Observable emit values is very different, in this particular case the fastest of both (crimes) was created from an Iterable and it's backpressure-enabled so it won't be an issue. But this is something to have in mind while merging different observables.
The next step is to filter the crimes we are interested in (burglaries). This can be done really easily by using the operator called filter (not a huge surprise, right?).
The filter operator emits only those items from an Observable that passes a predicate test. That is, given an Observable and a boolean function, filter will emit only the values for which the function returns true.
We want to filter crimes with code 310(BURGLARY) and 330(BURGLARY FROM VEHICLE).
We now have an Observable that emits exactly the data that we need, we will then use Spark micro framework to create a simple WEB showing the crimes as markers into a map.
We will use websockets for the communication between the backend and the frontend.
In the main function of our application we will setup Spark to expose all the files inside the public directory as static resources, we will also configure a redirection from / to /index.htmland lastly we will register a websocket handler using the /crimes path.
The websocket handler will simply keep a collection of active connections (sessions) that we are going to use in order to broadcast crimes as we process them.
On the client-side we will use Leaflet js and OpenStreetMap to create a map and show all the crimes published through the websocket by the backend.
The end result looks like this:
Conclusion
Even if the example shown in this article is contrived and uncomplicated, it's easy to see how RxJava provides a handy approach to represent the result of asynchronous computations in a way that is simple to use and composable.
If you require help in software development, you can contact us directly here. Otherwise you can find about all our services in our full website.
