Introduction to Reactive Programming

Reactive 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>() {

    @Override 

    public void onCompleted() { System.out.println("onCompleted called!"); }

    @Override

    public void onError(Throwable e) { System.out.println("onError called!"); }

    @Override

    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

}

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>() {

    @Override

    public void onCompleted() {

      System.out.println("May the force be with you!");

    }

    @Override

    public void onError(Throwable e) {

      throw new RuntimeException(e);

    }

    @Override

    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

• The Reactive Manifesto
• ReactiveX
• RxJava