RxJS is a library for dealing with streams.

Before we even start, here is the Stackblitz of the code I am using in this article

Observable

An Observable is a wrapper for dynamic data that allows ONE subscriber to be notified whenever the data changes.

Create an Observable

const elephant = Observable.create(observer => {
  observer.next('trump');
  observer.next('eyes');
  observer.complete();
  observer.next('tail'); // will not be broadcasted
});

elephant.subscribe(console.log);

const tiger = of('tiger');
hello.subscribe(console.log);// tiger

const giraphe = from('giraphe', asyncScheduler);
giraphe.subscribe(console.log); // g, i, r, a, p, h, e

const clickObservable = fromEvent(document, 'click');
clickObservable.subscribe(console.log);

const comingBackObservable = interval(500);
comingBackObservable.subscribe(console.log); // 0, 1, 2, 3 ...

Cold Observable

The default behavior : Whenever a new subscriber subscribes, the create method is played again.

Hot Observable

Hot Observables allows more than one subscriber to listen to the same Observable without repeating the call to the create method.

hot = cold.pipe(share());

The problem is that Hot Observables just mean “all subscribers subscribe to the same data source”. But the first subscriber still calls the create method and all future subscribers will be notified of FUTURE updates.

Hot Observable with replay

Most likely this is not the desired outcome because you still want to get the last emitted values when someone new subscribes. This is the job of “`shareReplay“` and “`publishReplay“`

hot=cold.pipe(shareReplay());
hot=cold.pipe(publishReplay());
hot=cold.pipe(publishReplay().refCount());

The (main?) difference between both is that publishReplay will automatically stop after the last subscriber unsubscribes if used with refCount.

nb: Don’t forget that you may want to subscribe later so shareReplay is a safer option as long as you don’t forget to close the stream manually.

Will I get all the last datas or just the previous one ?

Good question !

By default you will get ALL the datas from the stream (they have been stored in memory for you and all the future subscribers).

hot=cold.pipe(shareReplay(1));
hot=cold.pipe(publishReplay(1));

In most cases you just want the last one.

Subject

Most of the times you don’t want to create an hot observable from a cold one but create a Subject.

Subject behaves like hot observables (with the share method so they don’t store the history) but they do not have a create method, the messages are pushed .

BehaviorSubject

This is the holyGrail of Observables/Sujbects.

It is a Subject that works like it has a shareReplay(1) so it will give you the “current” value of the stream.

Close the pipes

unsubscribe

if you subscribe, then you should unsubscribe later

take()

take takes a number in agrument and closes the stream after.

takeWhile()

use the function to determine if it should close or not

takeUntil()

use an Observable to cancel the current stream when the Observable in parameters emits anything.

Operators

Pipe

Pipe is the function that you apply to a function in order to use operators on each value that “runs” in the operator.

const myNewObservable = myObservable.pipe(operator(), operator());

Pseudo code to use pipe

It is interesting to note that when a new value is emited, each observer will run through the stack of pipe for that value.

Deal with the datas

map()

map takes a function as argument, that function has the current value as a parameter and should return a new value that will replace it.

(oldValue) => {/* ... */return newValue}

tap()

tap is just like map but you can’t change the value by returning the new value. Basically it is used to DO something at that point of the pipe.

scan()

scan takes a function as argument, that function is like “reduce” in JS, it builds an accumulator (the last value) through the values.

(accumulator, oldvalue) => {return theNewValueThatWillBeTheNextAccumulator}

filter()

filter takes a function as argument, that function has the current value in parameter and should return true or false, only true values will be keps in the stream

(value) => {/* ... */return trueIfWeWantToKeepTheValue}

Too many data

debounceTime()

takes an number of ms in argument. It will filter out every value and if, after the number of ms in argument no value arrived, it will give the last value that passed.

Use case : search function where you don’t want to have a new value at every key stroke, but only a search when he stopped typing for some ms.

throttleTime()

takes an number of ms in argument. It will return AT MOST one value every “X ms”

bufferCount()

takes an integer and will aggregate chunk of that amount.

switchmap()

Switchmap is so important that it has it’s own category 🙂

Switchmap is a little like map but when you want to change the NATURE of the flow, it maps values to observable. Cancels the previous inner observable.

Use case : You have an Observable of the user and you want to use it’s ID to get a list of it’s posts.

posts$ = user$.pipe(switchMap(user => {return queryUserPostsReturnsObservable(user.ID))});

Combining Observables

The following functions

merge

Merge combines the Observables they receive as input parameters (spread) into a single Observable.

It just merge the observables and emits any new data from any source

combineLatest

Merge combines the Observables they receive as input parameters (in an array) into a single Observable.

From the moment there is some data in all the streams, whenever a new data comes in, the observer will receive all the latest data of all the streams.

Error handling

catchError

catches error and replace it with new value given as an Observable

retry

Automatically retries when error

Inspirations

http://reactivex.io/ -> https://rxjs.dev/