RXJava by Example

In the ongoing evolution of programming paradigms for simplifying concurrency under load, we have seen the adoption of java.util.concurrent, Akka streams, CompletableFuture, and frameworks like Netty. Most recently, reactive programming has been enjoying a burst of popularity thanks to its power and its robust tool set.

Reactive programming is a specification for dealing with asynchronous streams of data, providing tools for transforming and combining streams and for managing flow-control, making it easier to reason about your overall program design.

But easy it is not, and there is definitely a learning curve. For the mathematicians among us it is reminiscent of the leap from learning standard algebra with its scalar quantities, to linear algebra with its vectors, matrices, and tensors, essentially streams of data that are treated as a unit. Unlike traditional programming that considers objects, the fundamental unit of reactive reasoning is the stream of events. Events can come in the form of objects, data feeds, mouse movements, or even exceptions. The word “exception” expresses the traditional notion of an exceptional handling, as in - this is what is supposed to happen and here are the exceptions. In reactive, exceptions are first class citizens, treated every bit as such. Since streams are generally asynchronous, it doesn’t make sense to throw an exception, so instead any exception is passed as an event in the stream.

In this article we will consider the fundamentals of reactive programming, with a pedagogical eye on internalizing the important concepts.

First thing to keep in mind is that in reactive everything is a stream. Observable is the fundamental unit that wraps a stream. Streams can contain zero or more events, and may or may not complete, and may or may not issue an error. Once a stream completes or issues an error, it is essentially done, although there are tools for retrying or substituting different streams when an exception occurs.

Before you try out our examples, include the RxJava dependencies in your code base. You can load it from Maven using the dependency:

<dependency>
    <groupId>io.reactivex.rxjava</groupId>
    <artifactId>rxjava</artifactId>
    <version>1.1.10</version>
</dependency>

The Observable class has dozens of static factory methods and operators, each in a wide variety of flavors for generating new Observables, or for attaching them to processes of interest. Observables are immutable, so operators always produce a new Observable. To understand our code examples, let’s review the basic Observable operators that we'll be using in the code samples later in this article.

Observable.just produces an Observable that emits a single generic instance, followed by a complete. For example:

Observable.just("Howdy!")

Creates a new Observable that emits a single event before completing, the String “Howdy!”

You can assign that Observable to an Observable variable

Observable<String> hello = Observable.just("Howdy!");

But that by itself won’t get you very far, because just like the proverbial tree that falls in a forest, if nobody is around to hear it, it does not make a sound. An Observable must have a subscriber to do anything with the events it emits. Thankfully Java now has Lambdas, which allow us to express our observables in a concise declarative style:

Observable<String> howdy = Observable.just("Howdy!");
howdy.subscribe(System.out::println);

Which emits a gregarious "Howdy!"

Like all Observable methods, the just keyword is overloaded and so you can also say

Observable.just("Hello", "World")
          .subscribe(System.out::println);

Which outputs

Hello
World

just is overloaded for up to 10 input parameters. Notice the output is on two separate lines, indicating two separate output events.

Let’s try supplying a list and see what happens:

List<String> words = Arrays.asList(
 "the",
 "quick",
 "brown",
 "fox",
 "jumped",
 "over",
 "the",
 "lazy",
 "dog"
);

Observable.just(words)
          .subscribe(word->System.out.println(word));

This outputs an abrupt

[the, quick, brown, fox, jumped, over, the, lazy, dog]

We were expecting each word as a separate emission, but we got a single emission consisting of the whole list. To correct that, we invoke the more appropriate from method

Observable.from(words)
          .subscribe(System.out::println);

which converts an array or iterable to a series of events, one per element.

Executing that provides the more desirable multiline output:

the
quick
brown
fox
jumped
over
the
lazy
dog

It would be nice to get some numbering on that. Again, a job for observables.

Before we code that let’s investigate two operators, range and zip. range(i, n) creates a stream of n numbers starting with i. Our problem of adding numbering would be solved if we had a way to combine the range stream with our word stream.

RX Marbles is a great site for smoothing the reactive learning curve, in any language. The site features interactive JavaScript renderings for many of the reactive operations. Each uses the common “marbles” reactive idiom to depict one or more source streams and the result stream produced by the operator. Time passes from left to right, and events are represented by marbles. You can click and drag the source marbles to see how they affect the result.

A quick perusal produces the zip operation, just what the doctor ordered. Let’s look at the marble diagram to understand it better:

zip combines the elements of the source stream with the elements of a supplied stream, using a pairwise “zip” transformation mapping that you can supply in the form of a Lambda. When either of those streams completes, the zipped stream completes, so any remaining events from the other stream would be lost. zip accepts up to nine source streams and zip operations. There is a corresponding zipWith operator that zips a provided stream with the existing stream.

Coming back to our example. We can use range and zipWith to prepend our line numbers, using String.format as our zip transformation:

Observable.from(words)
 .zipWith(Observable.range(1, Integer.MAX_VALUE), 
    (string, count)->String.format("%2d. %s", count, string))
 .subscribe(System.out::println);

Which outputs:

 1. the
 2. quick
 3. brown
 4. fox
 5. jumped
 6. over
 7. the
 8. lazy
 9. dog

Looking good! Now let’s say we want to list not the words but the letters comprising those words. This is a job for flatMap, which takes the emissions (objects, collections, or arrays) from an Observable, and maps those elements to individual Observables, then flattens the emissions from all of those into a single Observable.

For our example we will use split to transform each word into an array of its comprising characters. We will then flatMap those to create a new Observable consisting of all of the characters of all of the words:

Observable.from(words)
 .flatMap(word -> Observable.from(word.split("")))
 .zipWith(Observable.range(1, Integer.MAX_VALUE),
   (string, count) -> String.format("%2d. %s", count, string))
 .subscribe(System.out::println);

That outputs

 1. t
 2. h
 3. e
 4. q
 5. u
 6. i
 7. c
 8. k
 ...
30. l
31. a
32. z
33. y
34. d
35. o
36. g

All words present and accounted for. But there’s too much data, we only want the distinct letters:

Observable.from(words)
 .flatMap(word -> Observable.from(word.split("")))
 .distinct()
 .zipWith(Observable.range(1, Integer.MAX_VALUE),
   (string, count) -> String.format("%2d. %s", count, string))
 .subscribe(System.out::println);

producing:

 1. t
 2. h
 3. e
 4. q
 5. u
 6. i
 7. c
 8. k
 9. b
10. r
11. o
12. w
13. n
14. f
15. x
16. j
17. m
18. p
19. d
20. v
21. l
22. a
23. z
24. y
25. g

As a child I was taught that our “quick brown fox” phrase contained every letter in the English alphabet, but we see there are only 25 not 26. Let’s sort them to help locate the missing one:

.flatMap(word -> Observable.from(word.split("")))
 .distinct()
 .sorted()
 .zipWith(Observable.range(1, Integer.MAX_VALUE),
   (string, count) -> String.format("%2d. %s", count, string))
 .subscribe(System.out::println);

That produces:

 1. a
 2. b
 3. c
 ...
17. q
18. r
19. t
20. u
21. v
22. w
23. x
24. y
25. z

Looks like letter 19 “s” is missing. Correcting that produces the expected output

List<String> words = Arrays.asList(
 "the",
 "quick",
 "brown",
 "fox",
 "jumped",
 "over",
 "the",
 "lazy",
 "dogs"
);

Observable.from(words)
 .flatMap(word -> Observable.from(word.split("")))
 .distinct()
 .sorted()
 .zipWith(Observable.range(1, Integer.MAX_VALUE),
   (string, count) -> String.format("%2d. %s", count, string))
 .subscribe(System.out::println);

 1. a
 2. b
 3. c
 4. d
 5. e
 6. f
 7. g
 8. h
 9. i
10. j
11. k
12. l
13. m
14. n
15. o
16. p
17. q
18. r
19. s
20. t
21. u
22. v
23. w
24. x
25. y
26. z

That’s a lot better!

But so far, this all looks very similar to Java Streams API introduced in Java 8. But the resemblance is strictly coincidental, because reactive adds so much more.

Java Streams and Lambda expressions were a valuable language addition, but in essence, they are, after all, a way to iterate collections and produce new collections. They are finite, static, and do not provide for reuse. Even when forked by the Stream parallel operator, they go off and do their own fork and join, and only return when done, leaving the program with little control. Reactive in contrast introduce the concepts of timing, throttling, and flow control, and they can attach to “infinite” processes that conceivably never end. The output is not a collection, but available for you to deal with, however you require.

Let’s take a look at some more marble diagrams to get a better picture.

The merge operator merges up to nine source streams into the final output, preserving order. There is no need to worry about race conditions, all events are “flattened” onto a single thread, including any exception and completion events.

The debounce operator treats all events within a specified time delay as a single event, emitting only the last in each such series:

You can see the difference in time between the top “1” and the bottom “1” as the time delay. In the group 2, 3, 4, 5, each element is coming within less than that time delay from the previous, so they are considered one and debounced away. If we move the “5” a little bit to the right out of the delay window, it starts a new debounce window:

One interesting operator is the dubiously named ambiguous operator amb.

amb is a conditional operator that selects the first stream to emit, from among all of its input streams, and sticks with that stream, ignoring all of the others. In the following, the second stream is the first to pump, so the result selects that stream and stays with it.

Sliding the “20” in the first stream over to the left makes the top stream the first producer, thereby producing an altered output:

This is useful for example if you have a process that needs to attach to a feed, perhaps reaching to several message topics or say Bloomberg and Reuters, and you don’t care which, you just need to get the first and stay with it.

Tick Tock

Now we have the tools to combine timed streams to produce a meaningful hybrid. In the next example we consider a feed that pumps every second during the week, but to save CPU only pumps every three seconds during the weekend. We can use that hybrid “metronome” to produce market data ticks at the desired rate.

First let’s create a boolean method that checks the current time and returns true for weekend and false for weekday:

private static boolean isSlowTickTime() {
 return LocalDate.now().getDayOfWeek() == DayOfWeek.SATURDAY || 
        LocalDate.now().getDayOfWeek() == DayOfWeek.SUNDAY;
}

For the purposes of those readers following along in an IDE, who may not want to wait until next weekend to see it working, you may substitute the following implementation, which ticks fast for 15 seconds and then slow for 15 seconds:

private static long start = System.currentTimeMillis();
public static Boolean isSlowTime() {
   return (System.currentTimeMillis() - start) % 30_000 >= 15_000;
}

Let’s create two Observables, fast and slow, then apply filtering to schedule and merge them.

We will use the Observable.interval operation, which generates a tick every specified number of time units (counting sequential Longs beginning with 0.)

Observable<Long> fast = Observable.interval(1, TimeUnit.SECONDS);
Observable<Long> slow = Observable.interval(3, TimeUnit.SECONDS);

fast will emit an event every second, slow will emit every three seconds. (We will ignore the Long value of the event, we are only interested in the timings.)

Now we can produce our syncopated clock by merging those two observables, applying a filter to each that tells the fast stream to tick on the weekdays (or for 15 seconds), and the slow one to tick on the weekends (or alternate 15 seconds).

Observable<Long> clock = Observable.merge(
       slow.filter(tick-> isSlowTickTime()),
       fast.filter(tick-> !isSlowTickTime())
);

Finally, let’s add a subscription to print the time. Launching this will print the system date and time according to our required schedule.

clock.subscribe(tick-> System.out.println(new Date()));

You will also need a keep alive to prevent this from exiting, so add a

Thread.sleep(60_000)

to the end of the method (and handle the InterruptedException).

Running that produces

Fri Sep 16 03:08:18 BST 2016
Fri Sep 16 03:08:19 BST 2016
Fri Sep 16 03:08:20 BST 2016
Fri Sep 16 03:08:21 BST 2016
Fri Sep 16 03:08:22 BST 2016
Fri Sep 16 03:08:23 BST 2016
Fri Sep 16 03:08:24 BST 2016
Fri Sep 16 03:08:25 BST 2016
Fri Sep 16 03:08:26 BST 2016
Fri Sep 16 03:08:27 BST 2016
Fri Sep 16 03:08:28 BST 2016
Fri Sep 16 03:08:29 BST 2016
Fri Sep 16 03:08:30 BST 2016
Fri Sep 16 03:08:31 BST 2016
Fri Sep 16 03:08:32 BST 2016
Fri Sep 16 03:08:35 BST 2016
Fri Sep 16 03:08:38 BST 2016
Fri Sep 16 03:08:41 BST 2016
Fri Sep 16 03:08:44 BST 2016
        . . .

You can see that the first 15 ticks are a second apart, followed by 15 seconds of ticks that are three seconds apart, in alternation as required.

Attaching to an existing feed

This is all very useful for creating Observables from scratch to pump static data. But how do you attach an Observable to an existing feed, so you can leverage the reactive flow control and stream manipulation strategies?

Cold and Hot Observables

Let’s make a brief digression to discuss the difference between cold and hot observables.

Cold observables are what we have been discussing until now. They provide static data, although timing may still be regulated. The distinguishing qualities of cold observables is that they only pump when there is a subscriber, and all subscribers receive the exact set of historical data, regardless of when they subscribe. Hot observables, in contrast, pump regardless of the number of subscribers, if any, and generally pump just the latest data to all subscribers (unless some caching strategy is applied.) Cold observables can be converted to hot by performing both of the following steps:

1. Call the Observable’s publish method to produce a new ConnectableObservable
2. Call the ConnectableObservable's connect method to start pumping.

To attach to an existing feed, you could (if you felt so inclined) add a listener to your feed that propagates ticks to subscribers by calling their onNext method on each tick. Your implementation would need to take care to ensure that each subscriber is still subscribed, or stop pumping to it, and would need to respect backpressure semantics. Thankfully all of that work is performed automatically by RxJava’s experimental AsyncEmitter. For our example, let’s assume we have a SomeFeed market data service that issues price ticks, and a SomeListener method that listens for those price ticks as well as lifecycle events. There is an implementation of these on GitHub if you’d like to try it at home.

Our feed accepts a listener, which supports the following API:

public void priceTick(PriceTick event);
public void error(Throwable throwable);

Our PriceTick has accessors for date, instrument, and price, and a method for signalling the last tick:

Let’s look at an example that connects an Observable to a live feed using an AsyncEmitter.

1       SomeFeed<PriceTick> feed = new SomeFeed<>(); 
2       Observable<PriceTick> obs = 
3         Observable.fromEmitter((AsyncEmitter<PriceTick> emitter) -> 
4         { 
5           SomeListener listener = new SomeListener() { 
6             @Override 
7             public void priceTick(PriceTick event) { 
8               emitter.onNext(event); 
9               if (event.isLast()) { 
10                emitter.onCompleted(); 
11              } 
12            } 
13   
14            @Override 
15            public void error(Throwable e) { 
16              emitter.onError(e); 
17            } 
18          }; 
19          feed.register(listener); 
20        }, AsyncEmitter.BackpressureMode.BUFFER); 
21   

This is taken almost verbatim from the Observable Javadoc; here is how it works - the AsyncEmitter wraps the steps of creating a listener (line 5) and registering to the service (line 19). Subscribers are automatically attached by the Observable. The events generated by the service are delegated to the emitter (line 8). Line 20 tells the Observer to buffer all notifications until they are consumed by a subscriber. Other backpressure choices are:

BackpressureMode.NONE to apply no backpressure. If the stream can’t keep up, may throw a MissingBackpressureException or IllegalStateException.

BackpressureMode.ERROR emits a MissingBackpressureException if the downstream can't keep up.

BackpressureMode.DROP Drops the incoming onNext value if the downstream can't keep up.

BackpressureMode.LATEST Keeps the latest onNext value and overwrites it with newer ones until the downstream can consume it.

All of this produces a cold observable. As with any cold observable, no ticks would be forthcoming until the first observer subscribes, and all subscribers would receive the same set of historical feeds, which is probably not what we want.

To convert this to a hot observable so that all subscribers receive all notifications as they occur in real time, we must call publish and connect, as described earlier:

22      ConnectableObservable<PriceTick> hotObservable = obs.publish(); 
23      hotObservable.connect(); 

Finally, we can subscribe and display our price ticks:
 

24      hotObservable.subscribe((priceTick) -> 
25        System.out.printf("%s %4s %6.2f%n", priceTick.getDate(), 
26          priceTick.getInstrument(), priceTick.getPrice())); 

About the Author

Victor Grazi is the Java queue lead at InfoQ. Inducted as an Oracle Java Champion in 2012, Victor works at Nomura Securities on core platform tools, and as a technical consultant and Java evangelist. He is also a frequent presenter at technical conferences. Victor hosts the "Java Concurrent Animated" and "Bytecode Explorer" open source projects.

Are you an architect or aspiring to be?

Stay on top of trends in the industry with one monthly newsletter, written by architects for architects.