Commonly Used Operators

In the last lesson, we covered the general idea behind the pipe method and operators. This lesson is going to focus on the most commonly used RxJS operators in detail. We will start with the more basic operators, and work our way up to the more complex ones.

As well as explaining the general idea behind each operator, I am also going to give you specific real world examples to demonstrate how they would actually be used - all of the examples will be from the example applications in this course, so you will also get the context for these later as well.

map, filter, tap

We already discussed how these three operators work in the last lesson, so we won’t rehash that again here. To quickly recap:

• map will allow you to transform values emitted on the stream by supplying a mapping function
• filter will allow you to prevent values from being emitted from the stream if they fail the provided predicate
• tap will allow you to use the value of the stream without modifying the stream - you can use this for debugging and side effects

The example we looked at in the previous lesson was this:

import { map, filter, tap } from "rxjs/operators"

myObservable$.pipe(
    tap((value) => console.log("Before map: ", value)),
    map((value) => value * 2),
    tap((value) => console.log("Before filter: ", value)),
    filter((value) => value < 7)
).subscribe((val) => console.log("Stream emitted:", val));

Which would result in the following output:

Before map:  1
Before filter:  2
Stream emitted: 2
Before map:  2
Before filter:  4
Stream emitted: 4
Before map:  3
Before filter:  6
Stream emitted: 6
Before map:  4
Before filter:  8
Before map:  5
Before filter:  10

Notably, the stream stops emitting data after the 6 value is emitted, because the predicate for the filter fails at that point.

One important thing to keep in mind is that the map and filter operators are different to the map and filter methods of an array. The map and filter operators apply to each individual stream emission. To demonstrate what I mean, imagine a stream that emits arrays of values:

const myObservable$ = of([1, 2, 3], [4, 5, 6], [7, 8, 9])

This would emit the following three arrays:

[1, 2, 3]
[4, 5, 6]
[7, 8, 9]

Now let’s say we want to double the values of all of the elements in these arrays. If you’re not careful, you might try to do something like this:

myObservable$.pipe(
    map((val) => val * 2)
)

But, this would result in the following:

NaN
NaN
NaN

We get Not a Number for all three emissions. Do you know why?

The map operator will map each individual stream emission. That means we are trying to do this:

[1, 2, 3] * 2
[4, 5, 6] * 2
[7, 8, 9] * 2

This doesn’t work. What we need to do is use both the map operator and the map array method. We use the map operator to modify the stream emission in some way, and the way in which we modify that stream emission is to use the map array method on it:

myObservable$.pipe(
    map((array) => array.map(val => val * 2))
)

This will give us the result we want:

[2, 4, 6]
[8, 10, 12]
[14, 16, 18]

The same goes filtering arrays that are emitted on a stream. We would do this to filter out odd values from data emissions:

myObservable$.pipe(
    map((array) => array.filter(val => val % 2 === 0))
)

Which would result in:

[ 2 ]
[ 4, 6 ]
[ 8 ]

Example use case

Although we have already talked about these three operators, we haven’t seen real world examples yet - let’s do that now:

  getChecklistById(id: string) {
    return this.getChecklists().pipe(
      filter((checklists) => checklists.length > 0), // don't emit if checklists haven't loaded yet
      map((checklists) => checklists.find((checklist) => checklist.id === id))
    );
  }

THe purpose of this method is to return a stream of a single checklist that matches the id passed in. We start with a stream of all checklists that the getChecklists() method returns, and the we pipe filter and map.

The map is a very typical use case. We map the emission which will be an array of all checklists, and then we use the standard find array method to return only the checklist that matches the id.

The usage of filter here is a little more creative. It is possible that getChecklists() will emit an empty array if the checklists have not been loaded into storage yet, which will result in no checklist being found by the map. To deal with this, the getChecklistById method will ignore any emissions from getChecklists that are just an empty array.

Imagine on a page in our application we are trying to get a specific checklist with getChecklistById and the data has not been loaded from storage yet. Without the filter our stream would emit a null value first which would cause us some trouble, and then it would emit the actual checklist after that. With the filter the first stream emission we get will be the checklist after it has loaded in from storage.

We will use a separate example for tap. The debugging use case is reasonably obvious for tap but let’s look at a case where we will use it to trigger a side effect:

getPhotos(){
    return this.photos$.pipe(
        tap((photos) => this.storage?.set('photos', photos))
    )
}

The photos$ stream will contain all of the photos that are present in the application. Every time a new photo is added (or deleted), this photos$ stream will emit. We are using tap here to trigger a side effect every time a new photo is added. We take the current photos and save them into storage.

startWith

The startWith operator is reasonably simple, let’s take a look at an example:

import { from } from "rxjs";
import { startWith } from "rxjs/operators"

const myObservable$ = from([1, 2, 3]).pipe(
    startWith(0)
)

This will cause the stream to emit:

0
1
2
3

The startWith will be subscribed to first, it will emit any values it was supplied, and then it will subscribe to the source observable (myObservable$) and emit all of its values.

Example use case

I find this to be particularly useful for streams that might not emit a value until some user interaction occurs. Sometimes, we need an initial value to kick things off, but if a stream doesn’t emit any values by default we can be stuck waiting.

This occurs quite often when we use the valueChanges observable from Angular’s ReactiveFormsModule:

const result$ = myFormControl.valueChanges.pipe(
    // do something
)

The valueChanges observable will emit every time the user changes a value in the form control. In one of the example applications, we let users view posts from a particular subreddit on Reddit. To let them specify the subreddit they want, we use a FormControl and we listen to valueChanges to set the subreddit appropriately.

However, valueChanges doesn’t emit anything by default. But, we still want to display a default subreddit. To handle this, we do something like this:

const result$ = myFormControl.valueChanges.pipe(
    startWith('gifs'),
    // fetch from subreddit with HTTP request
)

Now, even if the user has not entered anything into the form control, our stream will still emit a default value of gifs and kick off the process of fetching results from Reddit.

distinctUntilChanged

The distinctUntilChanged operator is another one that is reasonably simple, the basic idea is that it will prevent the stream from emitting the same value that was just emitted:

import { from } from "rxjs";
import { distinctUntilChanged } from "rxjs/operators"

const myObservable$ = from([1, 2, 3, 3, 2]).pipe(
    distinctUntilChanged()
)

This will cause the stream to emit:

1
2
3
2

Note that in the spot where we had 3, 3 only one 3 is emitted on the resulting stream. We can have the same values emitting multiple times on the stream (like the 2) as long as they are not immediately after each other.

Example use case