RxJS - Study Notes

RxJS is a reactive programming library for Javascript. It uses observable sequences to handle asynchronous operations and event handling. It provides core types such as Observable, satellite types (Observer, Schedulers, Subjects), and operators inspired by Array methods (map, filter, reduce, every, etc.) to allow handling of asynchronous events as collections.
ReactiveX combines the observer pattern with the iterator pattern and functional programming with collections to meet the ideal way of managing event sequences.
The basic concepts in RxJS for managing asynchronous events are:

Observable: Represents the idea of a callable collection of future values or events.
Observer: A collection of callbacks that knows how to listen to values delivered by the Observable.
Subscription: Represents the execution of the Observable, primarily used for canceling the execution.
Operators: Operators are pure functions that enable a functional programming style of handling collections with operations like map, filter, concat, reduce, etc.
Subject: Acts as both an Observable and an EventEmitter, and is the only way of multicasting a value or event to multiple Observers.
Schedulers: Schedulers are centralized dispatchers to control concurrency, allowing us to coordinate when computation happens on for example setTimeout or requestAnimationFrame or others.

Example
Usually, you register an event listener like this:

document.addEventListener('click', () => console.log('Clicked!'));

With RxJS, you can create an Observable like this:

import { fromEvent } from 'rxjs';

fromEvent(document, 'click').subscribe(() => console.log('Clicked!'));

Pure Functions#

What makes RxJS powerful is its ability to produce values with pure functions. This means your code is less prone to errors.
Usually, you would create an impure function where other parts of your code might mess with your state.

let count = 0;
document.addEventListener('click', () => console.log(`Clicked ${++counts} times`));

With RxJS, you can isolate the state.

import { fromEvent, scan } from 'rxjs';

fromEvent(document, 'click')
	.pipe(scan((count) => count + 1, 0))
	.subscribe((count) => console.log(`Clicked ${++count} times`));

The scan operator works similar to the reduce function on arrays. It takes an exposed value to the callback. The return value of the callback becomes the input for the next run of the callback.

Streams#

RxJS has a whole set of operators to help you control how events flow through your Observable.
Here's how you would allow at most one click per second using plain JavaScript:

let count = 0;
let rate = 1000;
let lastClick = Date.now() - rate;
document.addEventListener('click', () => {
	if (Date.now() - lastClick >= rate) {
		console.log(`Clicked ${++count} times`);
		lastClick = Date.now();
	}
});

With RxJS, you can achieve the same result:

import { fromEvent, throttleTime, scan } from 'rxjs';

fromEvent(document, 'click')
	.pipe(
		throttleTime(1000),
		scan((count) => count + 1, 0)
	)
	.subscribe((count) => console.log(`Clicked ${count} times`));

Other stream control operators include filter, delay, debounceTime, take, takeUntil, distinct, distinctUntilChanged, and more.

Values#

You can transform the values emitted by your Observable.
Here's how you would add the current mouse x position to each click in plain JavaScript:

let count = 0;
const rate = 1000;
let lastClick = Date.now() - rate;
document.addEventListener('click', (event) => {
	if (Date.now() - lastClick >= rate) {
		count += event.clientX;
		console.log(count);
		lastClick = Date.now();
	}
})

With RxJS, you can achieve the same result:

import { fromEvent, throttleTime, map, scan } from 'rxjs';

fromEvent(document, 'click')
	.pipe(
		throttleTime(1000),
		map((event) => event.clientX),
		scan((count, clientX) => count + clientX, 0)
	)
	.subscribe((count) => console.log(count));

Other value transformation operators include pluck, pairwise, sample, and more.