Development Losing My Marbles With Reactive Extensions

I was introduced to Reactive Extensions about 3 years ago, I first used them about 2.5 years ago, and I started to understand them about 2 years ago. And yes, those numbers are all correct.

Rx (I use Reactive Extensions, ReactiveX, and Rx interchangeably) is an interesting beast. One day I think I grok it, and the next day I’m left scratching my head trying to figure out if I should map, or flatMap. I think I’m clever with a combineLatest

• zip - and buffering that with a time-shift, and then I end up getting hit by a backpressure exception, even though my app has worked flawlessly for weeks.

Overall, my development life has improved immensely with the addition of reactive programming - specifically the Rx ports (as of writing, I have apps in the field running Rx.Net, RxJs, RxAndroid, RxPy - with RxSwift and RxCpp coming up in the next few months), but I’ll be damned if I don’t end up looking like a bush league developer every now and again.

What are Reactive Extensions?

The underlying premise of Reactive Extensions is that it is “a library for composing asynchronous and event-based programs by using observable sequences”, and moreover, from the ReactiveX website:

The Observer pattern done right ReactiveX is a combination of the best ideas from the Observer pattern, the Iterator pattern, and functional programming

While those explanations are accurate, I feel the spirit of Rx is better captured as something closer to inherently asynchronous programming with data streams. By ‘inherently’, I mean that if you do nothing special, you’re working in asynchronous mode, and you need to use specific APIs to work in synchronous mode (kind of like NodeJS).

ReactiveX is actually a collection of libraries for a large number of programming languages (so, I guess ReactiveX is better defined as an API or a concept). It originated as Reactive Extensions for the .NET platform, and has been ported to most mainstream languages since then. Quite often, certain companies take a lead on a language’s implementation and then the library is pulled into the ReactiveX Github repo. For example, I believe when I first started on it, RxJava was started by (or at least, heavily worked on and maintained by) Netflix - and then it was pulled into the ReactiveX brand.

Why Use Them?

I use Rx for two main reasons:

• The asynchronous data stream paradigm fits interacting with hardware better than callbacks

When I build hardware, I like it to be able to operate independently, rather than constantly waiting for instructions from an app.

For example, if you have a sensor - do you want to constantly poll data from it? Or have everyone who wants to see the data need to know how to talk to the sensor?

Or would you rather the sensor emits data, so that someone can listen in when they choose to - and if there is no one listening, the sensor will just know not to emit anything!

In case you’re wondering what the asynchronous and synchronous analogues are:

There are ways to emulate this same behaviour with callbacks and listener registrations, but that brings me to my next reason.

• Rx decouples code and reduces boilerplate

I think the following better explains this.

Rx comes baked in with all kinds of registration/notification mechanisms, stream operators, and best of all - it handles threading all on it’s own! Want to be on the UI thread? Just tell it to observeOn the main thread scheduler. Want a background thread? Call observeOn with a background scheduler (or subscribeOn, depending on what you’re doing). It’s as simple as that! Moreover, each observer can independently decide what operators/threads it wants, without affecting the operation of other observers.

Here is an example snippet in RxAndroid that tracks the connection state of a temperature sensor, and updates my UI (which MUST be done on the Main Thread):

mSubscriptions.add(TemperatureSensor.getInstance().getConnectionState()
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(state -> {
            mConnectionTextView.append(state.name() + ", ");
        }));

Required Reading

Rx can take a long time to grok, but there are a lot of excellent resources floating around.

• The ReactiveX Homepage - The place to start, and also where most of the documentation resides
• Intro to Rx - A .NET-centric walkthrough of Rx
• Grokking RxJava - An excellent blog series by Dan Lew on RxJava - he has simple explanations for much of the basic operation of Rx.
• RxMarbles - The visual explanation of what each Rx operator does using marble diagrams. For example, my favourite operator:

Final Thoughts

This was a bit of a fluff post, mostly to introduce the concept of Rx - and also provide some links to required reading.

In my next Rx-related posts, I will be covering implementation details about using RxAndroid (for example) and how to do some basic stream operations, explaining what subjects are, and so on. Currently, the beginning of a Github repo for that can be found here: https://github.com/sureshjoshi/rxandroid-examples

Feature Photo credit: Keenan Brennan / Foter / CC BY