RxJS Operators: What, Why and When ?

While writing some code in some front-end framework based on JavaScript/Typescript, we come across situations, where we need to perform some action or execute certain functionality based on some variable which keeps changing it’s value or keeps emitting something new every time.

The most common scenario is when we fetch or post some data asynchronously through HTTP to a server and based on the response of the request, we format that data and pass to a callback or render that in UI based on our requirement.

So to make all of that task easy for us, we use RxJS, as per the official definition:

The full form of RxJS is Reactive Extension for Javascript. It is a javascript library that uses observables to work with reactive programming that deals with asynchronous data calls, callbacks and event-based programs. RxJS can be used with other Javascript libraries and frameworks, it is defined as a library for composing asynchronous and event-based programs by using observable sequences. It provides one core type, the Observable, satellite types (Observer, Schedulers, Subjects) and operators inspired by Array#extras (map, filter, reduce, every, etc.) to allow handling asynchronous events as collections.

We all are aware of the Observables, they are commonly referred as lazy Push collections of multiple values. Which essentially means that Observables emit different values over a period of time. They are created only when we subscribe to them and emit the most recent values. To get the most of Observables or make our lives easier and our code more performant and efficient, we use various RxJS Operators.

Coming to RxJS Operators, “Operators are the essential pieces that allow complex asynchronous code to be easily composed in a declarative manner.” or

An Operator is a function which creates a new Observable based on the current Observable. This is a pure operation: the previous Observable stays unmodified.

An Operator takes and Observable as an input, performs some logic on the values streamed through the Observable, and creates a new Observable through these values without changing the original value

Operator Function Flow

We can see that the Operator takes in the values (Value1 and Value2) from Observable1, and creates a new Observable (Observable2), that is emitting new values (AlteredValue1 and AlteredValue2).

Let’s talk about 6 basic operators of RxJS, which we are going to use often while writing asynchronous operations:

1. of (Creation Operator)
2. from (Creation Operator)
3. map (Transformation Operator)
4. switchMap (Transformation Operator)
5. tap (Utility Operator)
6. take (Filtering Operator)

of: The of Operator is a creation Operator. Creation Operators are functions that create an Observable stream from a source.

The of Operator will create and Observable that emits different values in the sequence, followed by a Completion Notification. (A completion Notification tells the subscribers that the Observable will no longer emit new values).

Let’s take a look at how of works:

const arr = [1, 2, 3];

const arr$ = of(arr);

arr$.subscribe((values) => console.log(`Emitted Values: `, values));

of created the observable, and when we subscribe to it, it starts emitting the values immediately. The output of the this will be:

Emitted Values: [1, 2, 3]

of will emit the full array [1,2,3] as a full value.

from: The from operator turns an Array, Promise, or Iterable, into an Observable. This operator will convert a promise to an Observable, allowing it to be handled in a more reactive manner. When the Promise resolves or rejects, a complete notification will be sent to the subscriber.

The main difference between of and from is that from operator will emit each element in an array or Iterable in sequence, rather than full value. Once all the elements of the Array or Iterable have been emitted, a completion notification will be sent to subscribers.

Let’s look at how form works, by using the similar code we used for of operator:

const arr = [1, 2, 3];

const arr$ = from(arr);

arr$.subscribe((values) => console.log(`Emitted Values: `, values));

It’s output is:

Emitted Values:  1
Emitted Values:  2
Emitted Values:  3

Here we can see that, from Operator took each number and emitted it as a value. The subscriber received each value in sequence and called “console.log” 3 times.

map: The map operator is a Transformation Operator. It takes values from one Observable, transforms them, and creates an new Observable that emits the transformed values.

Let’s see how it works, we will take an Array, convert it to an Observable using from and modify is using the map:

const arr = [1, 2, 3];

const fromArr$ = from(arr);

fromArr$
  .pipe(map((value) => value + 10))
  .subscribe((value) => console.log(`Emitted Values: `, value));

You’ll notice the introduction of the .pipe() call. This is RxJS's method for applying operators to an Observable's stream before you subscribe to it. It will pipe the value emitted from the Observable through each operator passed as an argument, before passing the final transformed value to the subscribe method.

In this example, as map is a transformation operator, it must be used within pipe() call so it can transform the value it receives from the Observable.

The output of the above code is:

Emitted Values:  11
Emitted Values:  12
Emitted Values:  13

switchMap: It’s another transformation operator which means the it will piped through an Observable and then will modify or transform the resultant value. The switchMap receives the values emitted by an Observable, and then returns a new Observable from a different source.

Let’s understand this by an example:

Suppose you have an Observable that emits User IDs. You may want to fetch the full User object correlating to the ID, then do something with the full details. The switchMap operator would receive the ID from the Observable, then return an Observable containing the response from the request to fetch the User object.

I find it can be useful to think of this in the terms of switching streams. You're switching from one Observable stream to another.

const userDetails$ = from(this.userService.getActiveUserID())
    .pipe(switchMap(id => this.userService.fetchUserForID(id)))
    .subscribe(user => console.log("Found user ", user));

In the example, we ask for the active User’s Ids. Then, we ask the userService to fetch the user that correlates with the ID. We then subscribe to this new Observable stream, and receive the value it emits, rather than the values emitted from “from(this.userService.getActiveUserID())” as seen in the output:

Found user  {id: 1, name: "Test User", email: "test@test.com"}

It’s worth noting that the switchMap operator will cancel any in-flight network requests if it receives a new value from the original Observable stream (source), making it a great candidate for type ahead search implementations.

tap: The tap operator is a utility operator that returns an observable output that is identical to the source observable but performs a side effect for every emission on the source observable.

In other words, you can say that the RxJS tap() operator is used to intercept each emission on the source observable and runs a function but returns an output that is identical to the source observable as long as it doesn’t find any error.

tap allows you to perform actions or side effects on an Observable stream without modifying or altering the original stream. The values “pass-through” the tap operator to the next Operator or Subscriber.

This can be very useful for logging:

const arr = [1, 2, 3];

const fromArr$ = from(arr);

fromArr$
  .pipe(tap((value) => console.log("Received value: ", value)))
  .subscribe((value) => console.log(`Emitted Values: `, value));

The output:

Received value:  1
Emitted Values:  1

Received value:  2
Emitted Values:  2

Received value:  3
Emitted Values:  3

take: The take operator is a Filtering Operator. Filtering Operators allows you to select how and when to accept values emitted from the Observables.

take is one of the most common and most simplistic filtering operators. It allows you to specify a maximum number of values you want to receive from an Observable.

Let’s see how we can use it:

const arr = [1, 2, 3];

const fromArr$ = from(arr);

fromArr$
  .pipe(take(1))
  .subscribe((value) => console.log(`Emitted Values: `, value));

This will output:

Emitted Values:  1

Because we are taking only 1 emitted value in the take() operator.

It can be used in situations where we want to limit how many user-produced events (fromEvent) we want to handle, for example, the first time the user clicks in our app.

Happy Coding !!