Introduction to Reactive Programming
Or should I say: “Observable Pattern on steroids”
Reactive Programming is about dealing with data streams and propagation of change.
Reactive Systems are applications which architecture approach make them Responsive, Resilient, Elastic and Message Driven.
- Responsive: Systems should respond in a timely manner.
- Message Driven: Systems should use asynchronous message communication between components to ensure loose coupling.
- Elastic: Systems should stay responsive under high load.
- Resilient: Systems should stay responsive when some components fail.
According to the “Reactive Manifesto”:
“Reactive Systems are more flexible, loosely-coupled and scalable. This makes them easier to develop and amenable to change. They are significantly more tolerant of failure and when failure does occur they meet it with elegance rather than disaster. Reactive Systems are highly responsive, giving users effective interactive feedback.”
There are reactive libraries available for many programming languages that enables this programming paradigm.
Such libraries from the “ReactiveX” family are…
“..used for composing asynchronous and event-based programs by using observable sequences. It extends the observer pattern to support sequences of data or events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety, concurrent data structures, and non-blocking I/O.” (definition by ReativeX)
So, its possible to avoid the “callback hell” problem and abstract others issues concerning threads and low-level asynchronous computations. That makes our code more readable and focused in business logic.
“Ok, Ok… but how to do that???”
The Observable x Observer model
Simply put, Observable is any object that emits (stream of) events, that the Observer reacts to. The Observer object subscribes to an Observable to listen whatever items the Observable emits, so it gets notified when the Observable state changes. The Observer part is also called Subscriber or Reactor, depending on the library used.
The Observer stands ready to react appropriately when the Observable emits items in any point in time. This pattern facilitates concurrent operations because it doesn't need to block while waiting for the Observable to emit items.
The Observer contract expects the implementation of some subset of the following methods:
- OnNext: Whenever the Observable emits an event, this method is called on our Observer, which takes as parameter the object emitted so we can perform some action on it.
- OnCompleted: This method is called after the last call of onNext method, indicating that the sequence of events associated with an Observable is complete and it has not encountered any errors.
- OnError: This method is called when it has encountered some error to generate the expected data, like an unhandled exception.
Operators
Operator is a function that, for every element the source Observable emits, it applies that function to that item, and then emit the resulting element in another Observable.
So, operators operate on an Observable and return another Observable. This, way, operators can be combined one after other in a chain to create data flows operations on the events.
The behavior of each operator is usually illustrated in marble diagrams like this (Rx Marbles):
Reactive operators have many similarities to those of functional programming, bringing better (and faster) understanding of them.
Some of the most used core operators in ReactiveX libraries are:
- map: transforms items emitted by an Observable by applying a function to each them.
- flatMap: transforms the objects emitted by an Observable into Observables (“nested Observables”), then flatten the emissions from those into a single Observable.
- filter: emits only items from an Observable that pass a predicate test.
- just: converts objects into an Observable that emits those objects.
- takeWhile: discards items emitted by an Observable after a specified condition becomes false.
- distinct: suppress duplicate objects emitted by an Observable.
There is also an important concept of backpressure, which provides solutions when an Observable is emitting items more quickly than a Observer can consume them.
“Talk is cheap, show me the code!”
After some background theory, let's get to the funny part!
Below let's go through a hands-on approach, to provide an understanding by seeing the magic in motion! :-)
The examples use the RxJava (version 1.3.8) library:
<!-- maven dependency -->
<dependency>
<groupId>io.reactivex</groupId>
<artifactId>rxjava</artifactId>
<version>1.3.8</version>
</dependency>“Hello World!”
Here it is a simple inline “Hello World” code using Observable and immediate Subscription:
It's possible to do implicit or more explicit calls to Observer functions/methods:
The code above prints:
onNext => Message received: Hello World 4!
onCompleted called!Operators examples:
The output of the code above is:
4
8
12
16
20It's also possible to get an Observable from a List, a Callable or a Future instance:
Of course we can set <nerd mode on> and implement a Star Wars battle using Reactive Programming (source code here):
The output of the code above may be (troopers id numbers are random):
Jedi defeated Stormtrooper #246
May the force be with you!
Jedi defeated Stormtrooper #642
May the force be with you!
Jedi defeated Stormtrooper #800
May the force be with you!
Jedi defeated Stormtrooper #205
May the force be with you!Resources and further readings
