RxJS by Example: Part 2

Digging a bit deeper, we find that RxJS actually does not do much; Observables are more of a powerful code organization pattern.

This article is part of a series starting with RxJS by Example: Part 1.

lazy

It is important to observe that Observables, unlike Promises, are lazy, i.e., the function passed to create is not executed until it is subscribed to. For example the following code only outputs Hello from Promise.

src/lazy.js

/* eslint no-console: "off" */
import { Observable } from 'rxjs/Observable';
// eslint-disable-next-line
new Promise((resolve) => {
  console.log('Hello from Promise');
  resolve(1);
});
Observable.create((observer) => {
  console.log('Hello from Observable');
  observer.next(1);
  observer.next(2);
  observer.next(3);
});

not-emitter

My intuition (thinking of Promises) led me to misunderstand how Observables worked with multiple subscriptions (calls to subscribe). My guess was that the following code would output both Hello from Promise and Hello from Observable only once.

src/not-emitter.js

/* eslint no-console: "off" */
import { Observable } from 'rxjs/Observable';
// PROMISE
const myPromise = new Promise((resolve) => {
  console.log('Hello from Promise');
  resolve(1);
});
myPromise.then(o => console.log(o));
myPromise.then(o => console.log(o));
// OBSERVABLE
const myObservable = Observable.create((observer) => {
  console.log('Hello from Observable');
  observer.next(1);
  observer.next(2);
  observer.next(3);
});
myObservable.subscribe(o => console.log(o));
myObservable.subscribe(o => console.log(o));

Instead the output is:

Hello from Promise
Hello from Observable
1
2
3
Hello from Observable
1
2
3
1
1

My thinking was that similar to a Promise, I thought the function passed to create was only going to be called once and that the subscriptions just iterated through the results (like an event emitter might do).

It is important to observe that each subscription triggers a separate call to the function supplied to create.

Another subtle difference with Promises is that the subscribe happens synchronously whereas the then happens asynchronously (notice the two ones at the bottom of the output).

observer

Up until now we have been using shorthand notation; passing just a function to subscribe. The complete syntax allows you to pass an object (called an Observer) to subscribe with optional properties next, error, and complete. With this in mind, it now makes sense that we named the single parameter (observer) of the function supplied to create as we did.

src/observer.js

/* eslint no-console: "off" */
import { Observable } from 'rxjs/Observable';
const myObservable = Observable.create((observer) => {
  console.log('Hello');
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete();
});
myObservable.subscribe({
  next: o => console.log(o),
  error: err => console.log(err),
  complete: () => console.log('complete'),
});
myObservable.subscribe({
  next: o => console.log(o),
  error: err => console.log(err),
  complete: () => console.log('complete'),
});

note: Once the execution calls either complete or error any further calls to next, complete, or error are ignored.

note: You may have noticed that the Observer object passed to subscribe is not exactly the same object passed as the parameter of function supplied to create; otherwise, how could further calls be ignored. This is an example of the little bit of work that RxJS does; enforcing expected behavior.

subscription

The return value from the subscribe function is called a Subscription. The key feature of a Subscription is that it has a method, unsubscribe. Calling unsubscribe simply calls the function that is returned from the function supplied to create; in the following case clearing the interval. The output of the following function is (for five seconds output the word hi every second).

hi
hi
hi
hi

src/subscription

/* eslint no-console: "off" */
import { Observable } from 'rxjs/Observable';
const myObservable = Observable.create((observer) => {
  const intervalID = setInterval(() => {
    observer.next('hi');
  }, 1000);
  return () => {
    clearInterval(intervalID);
  };
});
const mySubscription = myObservable.subscribe(o => console.log(o));
setTimeout(() => mySubscription.unsubscribe(), 5000);

operators

At this point, I started to wonder what the whole point of Observables was; we are essentially simply passing an object (the Observer) to another function (create) and executing standard methods on it (next).

Like many other situations in programming, there is a surprising benefit of simply using standard patterns. In this case, by using Observables we can use RxJS operators to perform common tasks, e.g., map and filter that act much like the similarly named Array operators.

The output of the following code is:

MY_OBSERVABLE
1
2
3
4
MULTIPLY_OBSERVABLE
2
4
6
8
FILTER_OBSERVABLE
1
2

src/operators.js

/* eslint no-console: "off" */
import { Observable } from 'rxjs/Observable';
import 'rxjs/add/operator/map';
import 'rxjs/add/operator/filter';
const myObservable = Observable.create((observer) => {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.next(4);
});
const multiplyObservable = myObservable.map(o => o * 2);
const filterObservable = myObservable.filter(o => o < 3);
console.log('MY_OBSERVABLE');
myObservable.subscribe(o => console.log(o));
console.log('MULTIPLY_OBSERVABLE');
multiplyObservable.subscribe(o => console.log(o));
console.log('FILTER_OBSERVABLE');
filterObservable.subscribe(o => console.log(o));

note: RxJS provides many such operators.

Next Steps

In the next article, RxJS by Example: Part 3, we introduce additional key concepts; Subjects and Multicast Observables.