An introduction to Reactive Programming in JavaScript

This article was originally published in Spanish here.

Reactive programming is a declarative programming paradigm focused in (1) the usage of data streams and (2) change propagation. The zen of reactive programming is that everything is a stream.

But Osman, what is a stream?

About streams

According to Wikipedia:

In computer science, a stream is a sequence of data elements made available over time. A stream can be thought of as items on a conveyor belt being processed one at a time rather than in large batches.

Applying the zen of reactive programming: all the data and the events triggered during the lifecycle of an application can be represented as streams of data over time; and our programs are reactive when we react to these changes.

The most common example of reactiveness I can think of is the following:

const $increment = document.querySelector('#increment');
let counter = 0;

$increment.addEventListener('click', onIncrement);

function onIncrement(event) {
  counter += 1;
}

Yes, when registering onIncrement as an event listener for the click event, we are using reactive programming. We can think of every click on the $increment button as a collection of events over time, and for each event we react running onIncrement.

But not only clicks can be represented as a stream. Each time onIncrement is executed, we're mutating the value of counter over time, hence the value of counter can be represented as a stream as well.

Let's add a couple more functions to our example:

const $increment = document.querySelector('#increment');
const $decrement = document.querySelector('#decrement');
const $double = document.querySelector('#double');
const $halve = document.querySelector('#halve');

let counter = 0;

$increment.addEventListener('click', onIncrement);
$decrement.addEventListener('click', onDecrement);
$double.addEventListener('click', onDouble);
$halve.addEventListener('click', onHalve);

function onIncrement(e) {
  counter += 1;
}
function onDecrement(e) {
  counter -= 1;
}
function onDouble(e) {
  counter *= 2;
}
function onHalve(e) {
  counter /= 2;
}

The value of counter will change every time we click on any of the buttons and will be logged to the console. Unfortunately, changing the value of counter inside our event listener callbacks is not the best thing to do: even though we're expressing the intention of mutating counter in some way, it's not that clear that the value of counter is the result of a group of specific operations. It would be much better (or more declarative) if we could express the former code like this:

let counter = 0;

combine(
  onIncrementClick,
  onDecrementClick,
  onDoubleClick,
  onHalveClick,
).observe(result => {
  counter = result;
  console.log(result);
});

I'd dare to say, that even without knowing the implementation details of the combine and observe functions, it's way more clear and easier to understand the program's intention: from an initial value and combining a set of functions, we get a new value of counter.

Let's go back to the idea that everything is a stream.

The marble diagram above is a representation of click events on the button $increment over time, or phrasing it using the concepts we've been learning so far, a stream of clicks.

We can have a timeline for each button as well:

When combining our streams we have produced another stream called clicks. This stream is a representation of which buttons have been clicked over a window of time. If we associate each stream of clicks with a function that updates the value of counter, our clicks stream can also represent the value of counter over time.

We will see how to combine our events in a single stream later on.

Having clarified the nature of streams, let's talk about the second important concept in reactive programming, which is change propagation.

What is change propagation?

Change propagation is basically how we notify a module of our program that some of its dependencies have changed (or that some event has occurred) and that it has to react accordingly.

In reactive systems the ergonomics between the change emitter and whomever reacts to those changes has a different constitution than in proactive systems.

André Staltz explains this relationship brilliantly in this talk.

In summary, in a proactive system if a module triggers a change on another module, the former has the latter as a dependency. In reactive systems the relation is reversed. Let's see how this works with a code example:

class Cart {
  constructor(invoice) {
    this.items = [];
    this.invoice = invoice;
  }

  addItem(item) {
    this.items.push(item);
  }

  checkout() {
    const total = this.items.reduce(
      (acc, item) => acc + item.price,
      0
    );

    this.invoice.update(total);
  }
}

class Invoice {
  constructor() {
    this.total = 0;  
  }

  update(total) {
    this.total = total;
  }
}

const invoice = new Invoice();
const cart = new Cart(invoice);

cart.addItem({ name: 'Miller Lite', price: 1.99 });
cart.addItem({ name: 'Chips', price: 1.50 });
cart.checkout();

console.log(invoice.total); // 3.49

The Cart class has the Invoice as a dependency (we pass an instance of Invoice to Cart as a constructor argument). The moment we invoke the checkout method, the update method on Invoice is called directly. In this case, we're triggering a change on the class Invoice from Cart class, hence our system is proactive.

Let's take a look at the reactive version:

const EventEmitter = require('events');

class Cart {
  constructor() {
    this.items = [];
    this.emitter = new EventEmitter();
  }

  addItem(item) {
    this.items.push(item);
  }

  checkout() {
    const total = this.items.reduce(
      (_total, item) => _total + item.price,
      0
    );

    this.emitter.emit('checkout', total);
  }
}

class Invoice {
  constructor(cart) {
    this.total = 0;
    this.cart = cart;

    this.cart.emitter.on('checkout', this.update.bind(this));
  }

  update(total) {
    this.total = total;
  }
}

const cart = new Cart();
const invoice = new Invoice(cart);

cart.addItem({ name: 'Miller Lite', price: 1.99 });
cart.addItem({ name: 'Chips', price: 1.50 });
cart.checkout();

console.log(invoice.total); // 3.49

The main difference here is that now Invoice has Cart as dependency, the other way around from the former example. Now when we call the checkout method, an event called checkout is emitted with the total fee as payload. Notice that we have not updated the total value inside the Invoice instance, we have only emitted a message that the checkout event has occurred. The Invoice instance is listening for an event of type checkout and whenever this event is emitted, it handles it calling the update method.

This, ladies and gentlemen, is reactivity.

The Observer Pattern 👀

We already know the principles of reactive programming, however it would be a nice idea that we take a look at a design pattern that can help us modeling the interaction between different parts of our program: the Observer Pattern.

According to Wikipedia:

The observer pattern is a software design pattern in which an object, called the subject, maintains a list of its dependents, called observers, and notifies them automatically of any state changes, usually by calling one of their methods.

Let's take a look back to our first example:

const $increment = document.querySelector('#increment');
let counter = 0;

$increment.addEventListener('click', onIncrement);

function onIncrement(event) {
  counter += 1;
}

Let's say we want to have multiple counters instead of just one. How should we refactor our program? We could (a) modify the onIncrement function to update the counters inside its body, or (b) register more functions that listen to the same event.

Solution (a) is not that flexible. If we remove the event listener then we lose the ability to update any of our counters. We can't update one counter at the time as well:

const $increment = document.querySelector('#increment');
let counterA = 0;
let counterB = 0;

$increment.addEventListener('click', onIncrement);

function onIncrement(event) {
  counterA += 1;
  counterB += 1;
}

// After 3 seconds no counter will receive updates
setTimeout(() => {
  $increment.removeEventListener('click', onIncrement);
}, 3000);

For implementing solution (b) let's apply the Observer Pattern. We'll create a class called FromEvent that will serve as a subject:

class FromEvent {
  constructor(target, name) {
    this.observers = [];

    target.addEventListener(name, this.next.bind(this));
  }

  observe(observer) {
    this.observers.push(observer);
  }

  remove(observer) {
    this.observers = this.observers.filter(
      obs => obs !== observer
    );
  }

  next(e) {
    this.observers.forEach(obs => obs.next(e));
  } 
}

And we use our class FromEvent to register observers of the click event:

const $increment = document.querySelector('#increment');
let counterA = 0;
let counterB = 0;

const fromEvent = new FromEvent($increment, "click");

const updateCounter = (counter, callback) => (event) => {
  counter = callback(counter);
  console.log(counter);
};

const observerA = {
  next: updateCounter(counterA, value => value + 1)
};
const observerB = {
  next: updateCounter(counterB, value => value + 2)
};

fromEvent.observe(observerA);
fromEvent.observe(observerB);

// After 3 seconds only counterA will receive updates
setTimeout(() => {
  fromEvent.remove(observerB);
}, 3000);

In this case, we assign a high order function (updateCounter) to the next method of each observer, which is responsible for updating the value of each counter. Then we register each observer to the fromEvent subject using the observe function. This way, we have effectively model our program to have a source of changes (the subject) and multiple consumers (the observers).

Back to the beginning

With the knowledge we have acquired, let's implement our ideal API proposed at the beginning of the article:

let counter = 0;

combine(
  onIncrementClick,
  onDecrementClick,
  onDoubleClick,
  onHalveClick,
).observe(result => {
  counter = result;
  console.log(result);
});

We will start implementing combine, which is a function that takes two or more subjects as parameters and returns a new subject. The idea is that whenever any of the subjects emits an event, we must notify the observers of the subject returned by combine. When we register an observer on combine, we will also register that some observer on every of the subjects that we passed as arguments to combine.

const combine = (...subjects) => ({
  observe: observer => {
    subjects.forEach(subject => {
      subject.observe(observer)
    })
  }, 
});

And we can pass multiple streams as arguments:

const onIncrementClick = new FromEvent($increment, "click");
const onDecrementClick = new FromEvent($decrement, "click");

const clicks = combine(
  onIncrementClick,
  onDecrementClick
);

const clicksObserver = {
  next: e => console.log('click', e.target.id)
}

clicks.observe(clicksObserver);

Great! Now we have a way to combine multiple data streams.

Homework

So we covered all the bases about how to do reactive programming using the observer pattern. The only thing left to do now is map each stream of events to functions that can modify the counter value accordingly. In case you can't figure out where to start, you can add a map method to the FromEvent class in order to map each event to a value that's useful.

You can find my implementation of the solution here.

If you liked this article don't forget to share it and follow me on Twitter for more insights about JavaScript, Reactive Programming, Functional Programming, Frontend Development and more.

Cheers!