Reactive Programming: Dealing with Asynchronicity

Ricardo Sanz
sanzante / tunic
Decoupled Drupal Days 2018 @ New York

Ricardo Sanz
Drupal Developer & DevOps
drupal.org/u/tunic
sanzante @ Twitter

A way to manage asynchronicity.
This is NOT a talk about React.
Based on the ReactiveX library.
Usable in many languages in addition to JavaScript: PHP, Java, Scala, C#, C++, Pyhton and others.

It greatly simplifies asynchronous working.
Very easy to combine and manage asynchronous data sources.
Generates a shorter, simpler and more intuitive code^*.
Maintainability.

Unpredictable events over time.
The browser environment is asynchronous.
AJAX.
User interaction (mouse, keyboard, etc).
Timers or animations
WebSockets.
Workers.

Events that happen over time that we can't predict. This is very common in the browser because it's an asynchronous environment. If we examine a Drupal request we find that the exeuction path is more o less predictible (complex but predictible), and data is available when needed (when a database query is done we asume that result, success or failure, is available in the next line of code, the same for reading a file). In browser this is not true. We don't know when many operations will return their data: AJAX operations are the main example, but it's the same for workers, for example. User interaction, WebSockets, timers and animations are other sources of asynchronicty.

How we can manage this uncertainty, this asynchronicty?

Simple reaction to an event.

Function call (listeners).

               
jQuery('#grip').click(function() {
  jQuery('#panel').slideUp();
});

Coding complex reactions is hard.
Callback hell.

              
var doc;
memStorage.get(docId, response => {
doc = memResponse;
  if (!memResponse) {
    localStorage.get(docId, localResponse => {
      doc = localResponse;
      if (!localResponse) {
        serverStorage.get(docId, serverResponse => {
          doc = serverResponse;
          if (!serverResponse) {
            // Error.
          }
      });
    }
  });
});

Objects that wrap asynchronous operations and react when there is a result.
They're almost callbacks turned into objects.
They greatly improve flow and error management.
Added to ES2015 (ECMAScript 6).

They are resolved only once: they return a single value.
Following the execution flow may not be trivial.
They are not lazy.
They are not cancelables.
Promise hell.

If you want to repeat an action you must run the code that makes the action and processes the response of the promise again. Usually you can do it calling again the function that manages this, so it's not a big deal.. But you'll need to control this re-execution process.
This works well for managing setTimeouts or simple HTTP request responses, but falls short when dealing with setInterval or WebSockets, for example. Once the promise has been fullfilled no longer returns values. Get more info about this here: Promises vs Observables, by Mateusz Podlasin.
See article by Nolan Lawson, PouchDB core team developer: We have a problem with promises
As the promise is created the code that tries to solve it is executed immediately. It is assumed that the answer is needed now.
Except using some libraries like Bluebird, once the promise is created it will try to resolve or fail, but it cannot be cancelled (although it can be ignored, of course). This can result in wasted resources.

Although they are a great improvement over callbacks, the following strategy is even more powerful.

It brings functional programming capabilities to imperative programming.
Programming using asynchronous data streams.
Declaration of data pipes that process and transform the data that crosses them.
Declaration of reactions to changes.
Similar example: how Angular, React and Vue.js update rendering on data change.

Is about creating data pipes.
The pipe that certain data crosses.
Abstract idea, not a data structure ^*.

Example: Click event listener.

              
jQuery('#grip').click(function() {
  jQuery('#panel').slideUp();
});

Example: From a click to an AJAX request, and then to a instantiated object list.

Pipes with transformations are the interesting ones. For example, the transition from one click to a list of objects loaded from a service. At some point the user clicks a button. This click becomes a URL that serves, for example, a competitions's classification data list: athletes, points obtained, nationality, name, etc. That URL then becomes the data contained in that URL, which will be a JSON. Following that the JSON content is transformed into instances of a custom class that contains the data received. Lastly, the final action occurs, which is to assign these objects to an internal variable. If we use an Angular framework, it will happily paint those objects on the screen once they are assigned to the variable. So, the initial data, a click, has travelled a pipe that has been transforming that data, until the final action, the assignment to a variable. This is nothing new, it's done all the time using callbacks or promises. We could chain the output of one calling to another, and deal with the complexity that is produced. The good thing of Reactive Programming is how easy is to define this data management, using those defined pipes, creating that plumbing system.

Well, now we have to take a deep breath to get into the thick part of the talk. How do we create these data pipes? How do we declare the reactions to change?

Observables from Observer pattern.
Data streams.
They emmit data until they finish or an error is triggered.

Simple way to show data flow over time.
Data flow that emits 4 elements and then finishes.
Data flow that emits 3 elements and then emits an error.

They allow to chain streams.
They allow to transform the stream data.
They allow to modulate streams.
They can create streams.
They can combine streams

Filter operator, filters values that don't match the condition.

Map operator, transforms stream data.

Manual chronometre that accumulates time.
A button to enable it.
Another button to pause/continue.

https://codepen.io/sanzante/pen/gzyMNx

              
var clockValue = 0;
var clockOn = true;

const btS$ = Rx.Observable.fromEvent(buttonStart, 'click');
const btP$ = Rx.Observable.fromEvent(buttonPause, 'click');
const clock$ = Rx.Observable.interval(1000);

btP$.subscribe(() => clockOn = !clockOn);

clock$.subscribe(x => output1.innerHTML = x)

clock$.combineLatest(btS$, v => v)
  .filter(() => clockOn)
  .subscribe(() => output2.innerHTML = ++clockValue);

  clock$  ---1---2---3---4---5---6---7---8---9------>
                    do
          ---1---2---3---4---5---6---7---8---9------>
  btS$    --------x--------------------------------->
                  combineLatest
          --------2--3---4---5---6---7---8---9------>
                  filter (clockOn)
          --------2--3---4---6-----------8---9------>

  btP$    ---------------------x--------x----------->
  clockV  000000001112222333344444444444455556666--->
  clockOn 111111111111111111111000000000011111111--->

¡Button killer!

Detect clicks in a row (double, triple, etc).
Time is counted from the first click.

http://jsfiddle.net/okpjzsfh/5/

            
var button = document.querySelector('.this');
var clickStream = Rx.Observable.fromEvent(button, 'click');
var text = document.querySelector('h2');
var multiClickStream = clickStream
    .buffer(() => Rx.Observable.interval(660))
    .map(list => list.length)
    .filter(x => x >= 2);

multiClickStream.subscribe(numclicks => {
    text.appendChild(createItem(1));
});

multiClickStream.delay(3000)
    .subscribe(suggestion => {
      text.removeChild(h2elem.firstChild)
    });

            
Rx.Observable.fromEvent(button, 'click')
  .map(() => 'https://example.com/api/classification'
  .flatMap(url => http.get(url))
  .map(items => items.map(new MyItem(item)))
  .subscribe(instances => this.list = instances);

Auto refresh

            
Rx.Observable.fromEvent(button, 'click')
  .merge(Rx.Observable.interval(60000))
  .map(() => 'https://example.com/api/classification'
  .flatMap(url => http.get(url))
  .map(items => items.map(new MyItem(item)))
  .subscribe(instances => this.list = instances);

Deactivatable auto refresh

            
Rx.Observable.interval(60000)
  .filter(() => this.autorefresh)
  .merge(Rx.Observable.fromEvent(button, 'click'))
  .map(() => 'https://example.com/api/classification'
  .flatMap(url => http.get(url))
  .map(items => items.map(new MyItem(item)))
  .subscribe(instances => this.list = instances);

Deactivatable auto refresh and error management.

            
Rx.Observable.interval(60000)
  .filter(() => this.autorefresh)
  .merge(Rx.Observable.fromEvent(button, 'click'))
  .map(() => 'https://example.com/api/classification'
  .flatMap(url => http.get(url))
  .map(items => items.map(new MyItem(item)))
  .catch(err => Rx.Observable.of(this.list))
  .subscribe(instances => this.list = instances);

You can express from simple reactions to extremely complex reactions to change.
No more hells.
Observables resolve as many times as needed.
Execution flow is intuitive ^*.
It's lazy.
It's cancellable.

With many ashynchonous data sources.
Complex interactions.
Complex interdependency.
To avoid internal state to manage asynchronous events.

This is the mobile app fo the World Archery Federation, one of our clients.
You see three dropdowns, each is dependant on previous: if you change, the next dropdowns change.
Data is not just retrieved from and endpoint, we need some previous data to render sropdowns (identifiers, labels, etc), so we have dependant data. You could argue that API should be simpler to use and I would agree, but it wasn't the case and we had to stick to already existent API.
Data is refreshed, sometimes very often. While this view is pretty static, the underlaying library that access API is prepared to refresh data automatically and on demand. This would affect the displayed data.
App should work offline, so library builds a local data repository with the fetched data, requesting fresh data when it's outdated.
This resulted in a pretty asynchronous hell. We worked on this with AngularJS without reactive programming: we lost some mental health those days. Then we did it again with Angular using Reactive Programming, and although it was hard because we had to learn how to use it, the asynchronous hell just became a complex but funny problem to solve.

Get the application for Android or iOS.

They are not EventEmitter.
Can be asyncronous or synchronous.
Can be cold o hot.
Can be multicast o unicast.
Can share or not the data source with their subscriptors.

Because others have already done it.

It can express very complex models.
It handles any data source.
Observables: proposed for standard, currently in stage 1.
Less code, improved maintanibility.
Available in many languages.
Asynchronicity.

The introduction to Reactive Programming you've been missing: https://gist.github.com/staltz/868e7e9bc2a7b8c1f754
Interactive marble diagrams: http://rxmarbles.com/
Rx Visualizer: https://rxviz.com/
ReactiveX: http://reactivex.io/
Ben Lesh: https://medium.com/@benlesh/