RxJS – A Practical Intro to Operators

"Reactive programming is programming with asynchronous data streams."

Intro

When I first started learning about Angular, one thing I couldn’t avoid was RxJS. Coming from a promises, async/await background, I thought RxJS was a complete waste of time, and just couldn’t get my head around it or see what the benefit of it was; I felt using async/await was much cleaner.

Skip forward a few months and I think it just might be the best JS library on the internet!

One of the stumbling blocks for me was the lack of practical example available. Although the learnrxjs.io website provides plenty of examples for every operator, I always found it very hard to relate as all of their example consisted of creating observables strictly for the sake of demonstration. I’ve found that the most common use of RxJS comes with http requests. In this article, I will go through my most commonly used RxJS operators and how to use them. I will be using a public the api at: https://jsonplaceholder.typicode.com/todos to retrieve an array of todo items which we can then manipulate. I will use a result variable to store my result in the variable "result$", which I can then display in the html.

Map

The simplest of operators: take what you’ve been given, transform it in whatever way you want, then return something. Let’s say in our example, we had an api to retrieve all the todo’s, but we only want to display completed todo’s. Here, we would use the map operator:

this.result$ = this.httpClient
  .get("https://jsonplaceholder.typicode.com/todos")
  .pipe(map((x) => x.filter((x) => x.completed)));

Filter

This one really confused me at the beginning as I thought this would filter items from my observable array. What is actually will do is continue the stream IF the condition passes, hence it filters out certain streams.

For example, imagine in our scenario, we were only displaying todo’s for a certain user (the current user I guess), but what if the user is not logged in, or the user object doesn’t exist? In this case, we’d never was to display anything. We would add a filter to ensure the user is not null before continuing the stream:

this.result$ = this.httpClient
  .get("https://jsonplaceholder.typicode.com/todos")
  .pipe(filter((x) => this.user != null));

Tap

As simple as it sounds: tap (previously named do). Do something within the stream. I find this really useful for debugging by adding a quick console.log, but it has other good benefits too.

Say we wanted to display a loading spinner of some sort. Well, we could store in a variable whether we are currently loading, and tap at the end of our stream to signify that loading has finished:

this.loading = true;
this.result$ = this.httpClient
  .get("https://jsonplaceholder.typicode.com/todos")
  .pipe(tap((x) => (this.loading = false)));

combineLatest

One of my favourites. What it does: emits a value from the stream when either of the observables emit a value. This is good when you want to do something when ever more than 1 observable emits.

Say we wanted to add a search box for the todo’s. We can add a simple input like so:

<input [(ngModel)]="searchTerm" (ngModelChange)="search($event)" />

And each time the input value changes, we call the method "search" where we can update a variable "searchTerm$" I created:

searchTerm$: Subject = newSubject(); search(searchTerm: string) {    this.searchTerm$.next(searchTerm); }

Now for using combineLatest. We’ll combine both the value of the http.get request and the search term observable:

const allTodos$ = this.httpClient.get('https://jsonplaceholder.typicode.com/todos'); this.result$ = combineLatest(allTodos$, this.searchTerm$).pipe(map((obj: [ToDo[], string]) => obj[0].filter(x => x.title.indexOf(obj[1]) !== -1)));

So, if either observable changes, i.e. when you receive the todo’s from the api, or when you search, the list of todo’s displayed will be updated.

debounceTime

Another favourite of mine, very simple, yet very effective. Purpose: wait the specified time (milliseconds); if another value comes in before the specified time is up, take the new value and start waiting again, else if the time is up, the emit the value.

Scenario’s you’d want to use this is when you want to control user input; E.g. using autocomplete; if this is using an api, you don’t want to make a request every time the user types in a letter, but rather if they’ve stopped typing for 300ms or so.

Let’s apply this to our search example:

const allTodos$ = this.httpClient.get('https://jsonplaceholder.typicode.com/todos'); this.result$ = combineLatest(allTodos$, this.searchTerm$.pipe(debounceTime(500)), ).pipe(tap(x =>console.log(x)), map((obj: [ToDo[], string]) => obj[0].filter(x => x.title.indexOf(obj[1]) !== -1)));

Notice the debounceTime is piped onto the "searchTerm$" variable, and not our combineLatest or we just want to ‘slow down’ the search variable.

takeUntil

I find this very handy for unsubscribing all my subscriptions once the component is destroyed (nb. only required if you are actually subscribing with .subscribe()). The "takeUntil" takes an observable, and when this observable emits, the subscription will be destroyed, instead of keeping a reference for each subscription then unsubscribing in ngOnDestroy

Here’s an example:

destroy$: Subject = newSubject(); ngOnInit() {    const allTodos$ = this.httpClient.get('https://jsonplaceholder.typicode.com/todos').pipe(takeUntil(this.destroy$)).subscribe(); } ngOnDestry() {    this.destroy$.next(true);    this.destroy$.unsubscribe(); }

We’d want to do this is we weren’t either binding "allTodos$" in our html, or nothing else was observing it. Note, you need to use "takeUntil" for this to have the desired effect, using take (which takes a value as opposed to an observable), then the subscription will only be destroyed if the original observable emits (i.e. this.httpclient.get…).

Happy coding!