Reactive Programming — Using RxJS Operators

RxJS operators that can help simplify your web application development.

Source: Pixabay

I have been coding using the Reactive programming paradigm in my current gig — and I’m having a lot of fun with it. I already knew that this paradigm is based on the observer design pattern. However, I experienced a steep learning curve early on in my journey. Admittedly, after reading the documentation, my takeaway was how to create an Observable (Subject) and an Observer. I somehow got lost after reading through the operators.

The number of available operators you could use can be pretty overwhelming. The documentation is well-written. However, I think this is one of those concepts that you need to put into practice to get a deeper understanding of it. Or better yet, find existing problems that you can solve with RxJS operators.

In this post, I will walk you through the RxJS operators that I find helpful so far. The code samples will be using RxJS version 6.6.7 written in TypeScript.

You will notice that I have deliberately made my sample code as verbose as it can be:

• I am specifying types for Observables. Specifying types make it clear at what point we are initializing an Observable.
• There is an explicit use of return statements in arrow functions. It seems that there is a bug in JSFiddle where using the arrow function shortcut for return statement doesn’t work.
• There are variable assignments of observables that we might not need to do. However, I assigned some observables into a variable so that they’re easier to follow.
• I append the dollar sign ($) to Observables which is standard practice.

The basics of RxJS

Before we jump to the operators, let’s quickly review the basics.

You can skip this section if you already know how to use Observable, Observer and Subject in RxJS.

Observable and Observer

From theRxJS docs.

Observable: represents the idea of an invokable collection of future values or events.

Observer: is a collection of callbacks that knows how to listen to values delivered by the Observable.

Let’s start by creating anObservableinitialized with a collection of values and asetTimeoutfunction.

const observable$: rxjs.Observable = new rxjs.Observable(observer => { 
  observer.next(1); 
  observer.next(2); 
  observer.next(3); 
  observer.next(3.5); 
  setTimeout(() => { 
    observer.next(4); 
    observer.complete(); 
  }, 1000); 
});

Next, we will subscribe an observer to theObservable. The next call in the Observable will execute the callback function in our observer.

observable$.subscribe({ 
  next(x): void { 
    console.log('got value ' + x); 
  }, 
  error(err): void { 
    console.error('something wrong occurred: ' + err); 
  }, 
  complete(): void { 
    console.log('complete'); 
  }
});

When the observer has finished executing what’s in the observable collection, thecompletefunction is invoked.

The code sample will run as follows:

1. Prints “got value” — 1, 2, 3, and 3.5 in sequence.
2. Prints “just after subscribe.”
3. Prints “got value 4.”
4. Prints “Complete”

View Content

Code snippet by the author.

The code sample above gives us an idea about the concepts of observable and observers. In Reactive programming, anything can be a stream. Let’s explore a few examples of how we can implement these concepts in our Web UI.

Subject

To put it simply, aSubjectis a wrapper of theObservableand theObserver.

const subject$ = new rxjs.Subject();

We can conveniently create anObservableandObserverat once using theSubject. Then access them anywhere in our code through the instantiatedSubjectobject.

subject$.subscribe(x => console.log('Subscriber A: ' + x));
subject$.next(1);
subject$.next(2);

The code will print the following when we run it.

"Subscriber A: 1"
"Subscriber A: 2"

We can subscribe anotherObserverto ourObservablethat prints “Subscriber B.”

subject$.subscribe(x => console.log('Subscriber B: ' + x));
subject$.next(3);

We now have two observers available starting from this line.

"Subscriber A: 1"
"Subscriber A: 2"
"Subscriber A: 3"
"Subscriber B: 3"

If we want to prevent new subscriptions to our observable, we can use thecompletemethodto prevent new subscribe method calls silently.

subject$.complete();

Otherwise, we can useunsubscribeif we want to throw an error.

subject$.unsubscribe();

View Content

Code snippet by the author.

1. take

It takesnnumber of values out of the series of emitted values.

We can use take when we have a series of emitted values, and we want to take the firstnnumber ofemitted values only. This method is useful for capturing the first few click events or sampling from a list of values.

We have a series of emitted values 1, 2, 3, 4, and 5.

const emittedValues$: rxjs.Observable = rxjs.of(1, 2, 3, 4, 5);

We can use thetakeoperator to transform the observable’s emitted values by taking the firstnvalues.

Notice how we are passing ourtakeoperator to thepipemethod.

A Pipeable Operator is a function that takes an Observable as its input and returns another Observable. It is a pure operation: the previous Observable stays unmodified. —rxjs-dev

const takeValues$: rxjs.Observable = emittedValues$.pipe(rxjs.operators.take(2));

We can then subscribe to that observable with transformed emitted values.

takeValues$.subscribe(x => {
  console.log(x)
});

Our code will print the first two emitted values.

1
2

Experiment with the take operator using the sample code below.

View Content

Code snippet by the author.

2. filter

We can use the filter operator when we want to filter out the values emitted by our observable based on our set conditions. A simple example would be to filter out values that are even or odd numbers.

Let’s reuse the same emitted values in our previous example.

const emittedValues$: rxjs.Observable = rxjs.of(1, 2, 3, 4, 5);

We can use thefilteroperator with a callback that filters numbers withmodulo two equal to zero to filter out even numbers.

// filter even numbers
const evenNumbers$: rxjs.Observable = emittedValues$.pipe( 
  rxjs.operators.filter(x => {
    return x % 2 == 0
  })
);

console.log('even numbers');
evenNumbers$.subscribe(x => {
  console.log(x)
});

We will change the callback to filter odd numbers only —modulo two is not equal to zero.

// filter odd numbers
const oddNumbers$: rxjs.Observable = emittedValues$.pipe( 
  rxjs.operators.filter(x => {
    return x % 2 != 0
  })
);

console.log('odd numbers');
oddNumbers$.subscribe(x => {
  console.log(x)
});

Try other filter operator callbacks that you can think of.

View Content

Code snippet by the author.

3. map

Given a series of emitted values from an Observable, what if we want to append the string “take: ” for each emitted value? We can apply a function to each of the emitted values from the Observable using themapoperator.

const emittedValues$: rxjs.Observable = rxjs.of(1, 2, 3, 4, 5);

Let’s use a map operator function that takes a callback that appends the string “take: " to each value.

const mappedValues$: rxjs.Observable = emittedValues$.pipe( 
  rxjs.operators.map(x => {
    return 'take: ' + x
  })
);

Let’s subscribe a callback to our*$mappedValues*. The callback will print the values returned by our map function.

mappedValues$.subscribe(x => {
  console.log(x)
});

We will get the following output.

"take: 1"
"take: 2"
"take: 3"
"take: 4"
"take: 5"

Use other map operator callbacks that you can think of.

View Content

Code snippet by the author.

4. mergeMap

ThemergeMapoperator is useful when we want to work on two different Observables at once.

One specific example is if we want to do an API call to retrieve the User IDs and another API call to retrieve the User Profile information for each of the User IDs.

The mocked API call to retrieve User IDs.

// mock a userIDs API call
const getUserIds$: rxjsObservable = () => { 
  const userIds$: rxjs.Observable = rxjs.of(1, 2, 3, 4, 5); 
  return userIds$;
}

The mocked API call to retrieve User Profiles.

// mock a user profile API call
const getUserProfile$: rxjs.Observable = (userId) => { 
  const userProfiles$: rxjs.Observable = rxjs.of('a', 'b', 'c', 'd', 'e'); 
  return userProfiles$;
};

The API call that retrieves the User IDs will be our “Outer Observable,” The API call that retrieves the User Profile for each User ID will be our “Inner Observable.” We can apply this example to any dataset, not just Users, i.e., where you have to retrieve IDs or keys from an API and then use them or keys to retrieve another piece of information from another API.

const retrievedUsers$: rxjs.Observable = getUserIds$().pipe( 
  rxjs.operators.mergeMap(userId => { 
    return getUserProfile$(userId).pipe(
      rxjs.operators.map(userProfile => { 
        return userId + userProfile; 
      })
    ) 
  }), 
);

retrievedUsers$.subscribe(x => { 
  console.log('retrieved user: ' + x) 
});

From the example, we can see howmergeMaphelps combine two different resultsets or map the result of the outer Observable to the inner Observable.

Try it yourself with different datasets using the sample code below as a reference.

View Content

Code snippet by the author.

5. switchMap

The main difference betweenswitchMapand other flattening operators is the cancelling effect. On each emission the previous inner observable (the result of the function you supplied) is cancelled and the new observable is subscribed. You can remember this by the phraseswitch to a new observable.-learnrxjs.io

TheswitchMapoperator is helpful for typeahead implementations. For instance, we want to abandon the previous API request after the user enters a new character in the text input of our typeahead search.

We can demonstrate this by having a click event as our “Outer Observable” and anintervalas our “Inner Observable.” The click event observable will emit click events, well, of course, when it is clicked. The interval observable will emit a number that starts from zero every second.

const clickEvent$: rxjs.Observable = rxjs.fromEvent(document, 'click');

Taking ourclickEvent$as our “Outer Observable” and theintervalObservable as our “Inner Observable.”

rxjs.interval(1000)

This interval Observable will reset to zero because we switch to a new interval Observable every time we click the button. The “Outer Observable” is wraps our “Inner Observable.”

clickEvent$.pipe( 
  rxjs.operators.switchMap(() => { 
    return rxjs.interval(1000) 
  })
).subscribe(console.log);

The zeros in the output below indicate when the button is clicked; hence, it switches to a new interval Observable.

5601201

Try switching to different types of Observables using the sample code below.

View Content

Code snippet by the author.

Conclusion

There are a ton of other operators that you can learn. These are a few that I found helpful so far. The toolbox of operators I use might grow as I gradually use more of what RxJS will offer. Your learning journey will be different from mine, so I recommend trying out new operators you can find in the RxJS documentation.

More content atplainenglish.io