Introduction to Reactive Programming
This introduction to Reactive programming discusses the observable and oberserver model, as well as the operators and an example.
Join the DZone community and get the full member experience.
Join For FreeReactive programming is about dealing with data streams and the propagation of change.
Reactive systems are applications whose architectural 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 enable 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 patternto 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, it's possible to avoid the “callback hell” problem and abstract other issues concerning threads and low-level asynchronous computations. That makes our code more readable and focused in business logic.
The Observable x Observer Model
Simply put, an 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 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 the 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: suppresses 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.
“Show Me the Code!"
After some background theory, let's get to the fun 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>
Here it is a simple inline “Hello World” code using an observable and immediate subscription:
// simplest way (in-line)
public static void helloWorld1() {
Observable.just("Hello Reactive World 1!").subscribe(System.out::println);
}
It's possible to do implicit or more explicit calls to observer functions/methods:
xxxxxxxxxx
// explicit onNext function call
public static void helloWorld2() {
Observable<String> myObservable = Observable.just("Hello World 2!"); // not emitted yet
Action1<String> onNextFunction = msg -> System.out.println(msg);
myObservable.subscribe(onNextFunction); // item emitted at subscription time (cold observable)!
}
// explicit onNext and OnError functions call
public static void helloWorld3() {
Observable<String> myObservable = Observable.just("Hello World 3!"); // not emitted yet
Action1<String> onNextFunction = System.out::println;
Action1<Throwable> onErrorFunction = RuntimeException::new;
myObservable.subscribe(onNextFunction, onErrorFunction); // item emitted at subscription time (cold observable)!
}
Segregating Observable and Observer objects:
xxxxxxxxxx
public static void helloWorld4() {
Observable<String> myObservable = Observable.just("Hello World 4!");
Observer<String> myObserver = new Observer<String>() {
public void onCompleted() { System.out.println("onCompleted called!"); }
public void onError(Throwable e) { System.out.println("onError called!"); }
public void onNext(String msg) { System.out.println("onNext => Message received: " + msg); }
};
myObservable.subscribe(myObserver);
}
The code above prints:
xxxxxxxxxx
onNext => Message received: Hello World 4!
onCompleted called!
Since it is emitted just one item, it can be a single object:
xxxxxxxxxx
// since it is emitted just one item, it can be a Single object
public static void helloWorld5() {
Single<String> mySingle = Single.just("Hello World 5 from Single!");
mySingle.subscribe(System.out::println, RuntimeException::new);
}
Operators examples:
xxxxxxxxxx
public static void operatorsExamples() {
// filter = apply predicate, filtering numbers that are not even
Func1<Integer, Boolean> evenNumberFunc = x -> x%2 == 0;
// map = transform each elements emitted, double them in this case
Func1<Integer, Integer> doubleNumberFunc = x -> 2*x;
Observable<Integer> myObservable = Observable.range(1, 10) // emits int values from the range
.filter(evenNumberFunc)
.map(doubleNumberFunc);
myObservable.subscribe(System.out::println); // prints 4 8 12 16 20
}
The output of the code above is:
xxxxxxxxxx
4
8
12
16
20
xxxxxxxxxx
// Interval operator
private static void intervalExample() {
Observable.interval(2, TimeUnit.SECONDS) // emits a sequential number every 2 seconds
.take(5) // limit to first 5 elements
.toBlocking() // converts to a blocking observable
.subscribe(System.out::println); // prints 0 to 4 in 2 seconds interval
}
It's also possible to get an Observable from a List , a Callable or a Future instance:
xxxxxxxxxx
// Creating Observables from a Collection/List
private static void observableFromListExample() {
List<Integer> intList = IntStream.rangeClosed(1, 10).mapToObj(Integer::new).collect(Collectors.toList());
Observable.from(intList).subscribe(System.out::println); // prints from 1 to 10:
}
// Creating Observables from Callable function
private static void observableFromCallableExample() {
Callable<String> callable = () -> "From Callable";
// defers the callable execution until subscription time
Observable.fromCallable(callable).subscribe(System.out::println);
}
// Creating Observables from Future instances
private static void observableFromFutureExample() {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "From Future");
// converts a Future into Observable
Observable.from(future).subscribe(System.out::println);
}
Of course, we can set <nerd mode on> and implement a Star Wars battle using Reactive Programming (source code here):
xxxxxxxxxx
// StarWar battle: Let's get nerdy...
private static void starWarsBattle() {
Random random = new Random();
// Stormtrooper class
class Stormtrooper {
private int imperialNr;
public Stormtrooper(int imperialNumber) {
this.imperialNr = imperialNumber;
}
public String getName() {
return "#" + imperialNr;
}
}
// callable func that creates a Stormtroper after 3 seconds delay
Callable<Stormtrooper> trooperGenerator = () -> {
Thread.sleep(3 * 1000);
return new Stormtrooper(random.nextInt());
};
// Creating Observables of Stormtrooper creation
List<Observable<Stormtrooper>> observables = IntStream.rangeClosed(1, 4)
.mapToObj(n -> Observable.fromCallable(trooperGenerator)).collect(Collectors.toList());
// Jedi observer to fight every tropper created in time
Observer<Stormtrooper> jedi = new Observer<Stormtrooper>() {
public void onCompleted() {
System.out.println("May the force be with you!");
}
public void onError(Throwable e) {
throw new RuntimeException(e);
}
public void onNext(Stormtrooper t) {
System.out.println("Jedi defeated Stormtrooper " + t.getName());
}
};
// Jedi subscribe to listen to every Stormtrooper creation event
observables.forEach(o -> o.subscribe(jedi)); // Battle inits at subscription time
}
The output of the code above may be (troopers ID numbers are random):
xxxxxxxxxx
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
Published at DZone with permission of Tiago Albuquerque. See the original article here.
Opinions expressed by DZone contributors are their own.
Comments