RxJS Quick Introduction (with Examples) for Beginner-Level Angular Developers

Table of contents:

1. Before we start… 
2. Using http client to get backend data straight into Angular template
3. Getting list of ids/hashes at being just to get list of full models later
4. Handling search results and social posts interactions in RxJS    
5. What’s next? Closing remarks

Requirements

Basic knowledge about JavaScript/TypeScript/Angular. 

Before we stat with RxJS Angular 

I need to mention that all code from this “quick introduction” is available on https://github.com/codete/codete-rxjs-quick-start

RxJS is a fantastic tool to reduce mess in your JavaScript application (and not only!). In this post, I want to cover basic things that will help you in ~90% of your Angular/RxJS work… Pretty good deal, right?

For any JavaScript development I recommend using Chrome Developer Tools to see clearly interactions between browser and server api.  

How to access Developer Tools in Chrome

Questions that you probably may have

Q: What about promises ? Will I never use them again or should I still use them?
A: It depends… For your professional work with RxJS/Angular, 99% of the time you can use observables + also you can create observables from promise (although you don’t need to do that because in the new Angular with http client you have everything you have ever wanted). 
 

Q: What are the observables?

A: Observables are the primary type in RxJS. They are similar to promises, but with extra features like multicast, cancelation, streams operators.

Q: Can I use async/await syntax from JavaScript to get async data?
A: Generally, yes... But there is a better way to do it! Your specific async/await  place in code can be replaced with observables merged by ex.  mergeMap(), combineLatest() - more on this topic later.

Q: What about Promise.all()? Is there something similar?
A: Sure! There is forkJoin() operator and a lot more…

Q: What is the main advantage of observables over promises in regards to http requests?
A: Observables can be “canceled” any time and it makes them perfect for handling (or canceling) many http requests at the time. For sure there are libraries for promises with “cancel” feature, but ES6 built-in promises are still without this possibility… and without many brilliant ideas from the RxJS team/community.

Little comparison between promises and observables

*Hot observables already have a producer (before “subscribe” action) in contrast to cold observables, which create a producer during first use (actually, when something is using the .subscribe() function on an observable for the first time). “Producer” here means: something that produces values.

Simplest example of a hot observable is an observable created from mouse movement (before creating this observable we are moving mouse for sure) and a cold observable - is an observable created from a http request (which first requires a new instance of requests producer XMLHttpRequest, that should not be reused). More info about hot/cold observables later in this post.

Using http client to get backend data straight into Angular template 

Things covered in this paragraph:

• async pipe in Angular template vs  pipe() function for RxJS
• operators:, map, tap operators
• reusing observables, share() operator

Simple array of authors with async pipes in Angular template 

When you start learning a new technology, it is important that at the beginning you are excited about it and you can use it to your own advantage asap. In my quick tutorial, I will try to explain as simply as possible things that are happening behind the scenes of RxJS cooperating with Angular.

The very first example that I want to show you (my new RxJS ninja) is a simple frontend-backend app for injecting data array into Angular template. The key word that you need to be familiar with here is an “observable”.  Observables, as a core part of RxJS, show the most elegant way of dealing with complex/asynchronous data in the browser/JavaScript. 

backend-data-into-template.component.ts

authors$ = this.http.get<Author[]>('/api/authors')

backend-data-into-template.component.html

Authors:
<mat-card>
 <ul>
   <li *ngFor="let author of (authors$ | async)">
     {{author.name}}
   </li>
 </ul>
</mat-card>

Array of authors displayed inside template

In the example above, variable authors$ gets a stream of data from http request. Once data is in a form of observable, we can proceed with our stream manipulation/merging/chaining etc. (in this case, we are only displaying raw data). 

In RxJS, it is crucial to understand the concepts of Producer and Consumer. Angular template here is a Consumer - it subscribes (underhood calls method .subscribe() on observable variable) to authors$ and, every time it does this, it creates a Producer (https request) to get data into observable stream. Observables that we use to produce data at the moment of subscribing are called Cold observables. Just keep that in mind for now - you will see later why it is so important.

api-interfaces.ts

export interface Author {
 id: number;
 name: string;
 books: (Book | number)[];
}

app.service.ts

const authors: Author[] = [
 {
   id: 1000,
   name: 'Elon',
   books: [1_000_000, 1_000_001]
 },
 {
   id: 1000,
   name: 'Charles',
   books: [1_000_002]
 },
 {
   id: 1000,
   name: 'Donald',
   books: [1_000_003]
 },
];

To present data from any observable (hot or cold) directive “ngFor” (from version 4+ of Angular) has become an amazing tool for presenting any kind of data streams that are observable-based. Therefore, populating data into an Angular template is now super simple - as you can see in the example above backend-data-into-template.component.html.

async pipe & .pipe() function

To achieve this simplicity, Angular’s core team implemented async pipe (in html template, fragment “| async''), which is responsible for resolving values from observables into normal data and now it is a part of “Angular magic” that lots of developers love. 

On the other hand, in RxJS we also have something called pipe, but it is a totally different pipe. RxJS .pipe() function is our “entrance” to the “world” of stream manipulations through RxJS operators and I am using it here with authors$ observable (share() operator) to create observable authorsWithAtLeastOneBook$ - this way, I prevent same https requests from multiplying.
 

backend-data-into-template.component.ts

authors$ = this.http.get<Author[]>('/api/authors').pipe( share() )

backend-data-into-template.component.ts

authorsWithAtLeastOneBook$ = this.authors$.pipe(
   . . .

Solution for multiple subscriptions of the same observable - share() operator

Operator share() “shares” stream between many subscribers asking for the same thing. This also applies when you use the same observable variable many times inside an Angular template.

Results of adding share() operator

authors$ = this.http .get<Author[]>('/api/authors') .pipe( share() )	authors$ = this.http .get<Author[]>('/api/authors')

Array of books with “loading” template as “else” option for async pipe load time

Now when we know how to easily display data from the backend in template - we can add a simple loading message to indicate to the user that we are actually doing something. Let’s have a look on books$ observable:
 

backend-data-into-template.component.ts

books$ = this.http.get<Book[]>('/api/books')

backend-data-into-template.component.html

<mat-card>
 <ul *ngIf="books$ | async; else elseLoadingTemplate; let books">
   <li *ngFor="let book of books">
     {{book.title}}
   </li>
 </ul>
</mat-card>
<ng-template #elseLoadingTemplate>
 loading books...
</ng-template>

In “real world”, this code probably would do a pretty good job, but in our local instance of serve, we are getting server responses too quickly to see our loading… Let's fix that with a delay() operator that is going to add 1 second (1000ms) before any backend (observable stream) data appears.

backend-data-into-template.component.ts

books$ = this.http.get<Book[]>('/api/books').pipe(
   delay(1000)
 );

Books loader for books$ observable inside Angular template in loading state

Use of map() and tap() operators to filter/debug authors with at least 2 books

Usually the json that we get from the backend isn’t prepared exactly how we want and we need to do some data modification. In RxJS api, the best/simplest thing to achieve this comes with map() operator.
 

backend-data-into-template.component.ts

authorsWithAtLeastOneBook$ = this.authors$.pipe(
   map(authors => {
     return authors.filter(({ books }) => books.length >= 2)
   })
 );

This operator, taken from ES6 into RxJS, allows us to display a filtered list of authors with at least 2 books - or whatever we want to do with stream data - before it goes into template.

Having more operators like map() is super convenient. We have everything inside one variable and we don’t need to use components classes variables as much. The only thing that may concern us is  the debugging process, which actually becomes a little more complex. To solve this problem, we can add a tap() operator wherever we think it makes sense (with console.log) to display stream state more accurately and without any modification. 

 

backend-data-into-template.component.ts

 authorsWithAtLeastOneBook$ = this.authors$.pipe(
   map(authors => {
     return authors.filter(({ books }) => books.length >= 2)
   }),
   tap(console.log)
 );

Results of all code from this paragraph... combined.

Tap() operator with console.log as way to debug values between operations in .pipe(..) function

Getting list of ids/hashes at being just to get list of full models later 

Things covered in this paragraph

• merging observables streams (mergeMap, concatMap)
• combineLatest,combineLatestWith

Getting list of posts (numbers array) and getting content for each next post - use of operators: mergeMap, concatMap, combineLatest,combineLatestWith

Hamster album is an app that combines lots of basic things from RxJS which are necessary to deal with more complex cases and make it super easy. Data here has a similar structure to previous authors/books entities, with hamsters owners as a main object that contains ids of hamsters to resolve + also, I am keeping api requests in a separated service (which is something definitely recommended).
 

list-of-ids-to-full-models.service.ts

constructor(private http: HttpClient) { }

 getHamsterOwners() {
   return this.http.get<HamsterOwner[]>('/api/hamsters/owners')
 }

 getHamsterById(id: number) {
   return this.http.get<Hamster>(`/api/hamster/${id}`)
 }

app.service.ts

export type Grams = number;

export interface Hamster {
 name: string;
 weight: Grams;
 id: number;
}

export interface HamsterOwner {
 name: string;
 hamsters: ( number | Hamster)[];
}

export interface HamsterOwnerFull extends Omit<HamsterOwner, 'hamsters'> {
 hamsters: Hamster[];
}

app.service.ts

export const Hamsters: Hamster[] = [
 {
   id: 1000,
   name: 'Pluszka',
   weight: 90
 },
 {
   id: 1001,
   name: 'Misia',
   weight: 70
 },
 {
   id: 1002,
   name: 'Chrupka',
   weight: 90
 },
 {
   id: 1003,
   name: 'Łapczuch',
   weight: 60
 }
];

app.service.ts

export const HamstersOwners: HamsterOwner[] = [
 {
   name: 'Megan',
   hamsters: [1002, 1003]
 },
 {
   name: 'Natalie',
   hamsters: [1000]
 },
 {
   name: 'Victoria',
   hamsters: [1001]
 }
]

The component part of the hamsters album app contains 3 observables. The first one - owners$ - is used here for getting data from the backend and then sharing the results to the next 2 observables.

This way, I am keeping all logic inside observables. Generally, you should keep your RxJS/Angular like this - to have great consistency/readability across your components.
 

list-of-ids-to-full-models.component.ts

owners$ = this.api.getHamsterOwners().pipe(share())
 

list-of-ids-to-full-models.component.ts

allHamsters$ = this.owners$.pipe(
   map((arr) => {
     const res = arr.reduce((a, b) => {
       return a.concat(b.hamsters as any);
     }, [] as number[])
     return res;
   }),
   mergeMap(arr => { // concatMap is OK here also
     const obs = arr.map(id => this.api.getHamsterById(id));
     return combineLatest(obs);
   }),
   share()
 );

Operators mergeMap() and combineLatest() are used here for resolving data (similarly to “await/async” in “for” loop from JavaScript) based on some previous resolved id-s (identificators). Also, instead of using the mergeMap() operator, we can use concatMap() if order of request does matter (in the example above, it doesn’t, as we want to get all data as quickly as possible – in no particular order).
 

list-of-ids-to-full-models.component.ts

ownersWithHamsters$ = this.owners$.pipe(
   combineLatestWith(this.allHamsters$),
   map(([owners, hamsters]) => {
     return owners.map(a => {
       a.hamsters = a.hamsters.map(hamsterId => hamsters.find(({ id }) => id === hamsterId)) as any;
       return a as HamsterOwnerFull;
     });
   })
 );

list-of-ids-to-full-models.component.html

<h2>
 All hamsters
</h2>
<mat-card *ngFor="let hamster of (allHamsters$ | async)">
 <app-hamster [hamster]="hamster"></app-hamster>
</mat-card>

<h2>
 Owners and hamster
</h2>
<mat-card *ngFor="let owner of (ownersWithHamsters$ | async)">
 <h3> {{ owner.name }} </h3>
 <h4>owned hamsters:</h4>
 <ul>
   <li *ngFor="let hamster of owner.hamsters">
     <app-hamster [hamster]="hamster"></app-hamster>
   </li>
 </ul>
</mat-card>

Observable ownersWithHamsters$ uses both previously defined owners$ and $allHamster to “combine latest data” (with combineLatestWith() operator) from both streams and replace hamsters ids with real, full models.  

In “real life” scenarios, similar situations from this example happen all the time. Like, for instance, “posts with comments ids”, “users with friends ids” etc. - I do recommend creating something similar to this to practice RxJS “stream combining” thinking. Apart from already mentioned operators from this paragraph, a couple of operators also worth noting are: forkJoin(), withLatestFrom() and merge() - they are similar in some ways.
 

Hamsters album app

Handling search results and social posts interactions in RxJS Angular 

Things covered in this paragraph:
- operators: switchMap, fromEvent, of, defer, scan, filter, catchError, EMPTY, exhaustMap, takeUntil

Simple hamster search (switchMap, debounce), word counter (filter,reduce)

RxJS is a fantastic library and once you get used to doing everything with it, your code becomes more and more consistent and readable. In the next example of this tutorial, I would like you to show a hamster search app (for finding your best furry friend) that will make you go back to this tutorial many times.

First look at the hamsters search

Search functionality is an extremely common feature in practically every web app. To build it with RxJS/Angular, we have a switchMap() operator. This operator cancels requests when you type/produce too many of them. By canceling requests, you are not only saving browser/client cpu power, but also actual server resources. 

Btw, in this example I am using hot observables. This means that the Producer of data existed before we subscribed to observable (means: user input was already there before we subscribed) and probably it is emitting data constantly (cold observable completes with 1 result).

Search input is the main component of this app and can be implemented in many ways - through Angular api or through RxJS operators like fromEvent(). I’ve chosen fromEvent(), because I want to operate only on OBSERVABLES as, this way, everything inside the component can be understood more quickly + also,  I will also use fromEvent() with defer() to save some memory/cpu and create an observable that exists only when anyone subscribes to it.

hamster-search.component.html

 <mat-label>Type name of your little, chubby friend</mat-label>
 <input matInput
        #search>

hamster-search.component.ts

 @ViewChild('search', { static: true }) search?: ElementRef<any>;
 searchInputChange$ = defer(() => fromEvent<KeyboardEventType>(
this.search?.nativeElement as any, 'keyup'))
   .pipe(
     map(c => c.target.value),
     share(),
   );

hamster-search.component.ts

  @ViewChild('search', { static: true }) search?: ElementRef<any>;
 searchInputChange$ = defer(() => fromEvent<KeyboardEventType>(
this.search?.nativeElement as any, 'keyup'))
   .pipe(
     map(c => c.target.value),
     share(),
   );

For the purpose of building an app where multiple requests don’t “care” about errors, I’ve used the catchError() operator, which returns an empty observable once something goes wrong. This way, the process of searching can be continuous.

Also, in the api service, I’ve prevented a situation when we try to search with an empty string - by using operator of() to create a dummy, empty result. Here the of() operator doesn’t do anything special, but usually it is a very useful thing to create cold observables from arrays or parameters (just like interval() for hot observables ).
 

hamster-search.service.ts

getHamsterByName(name: string) {
   if (!name) {
     return of([]);
   }
   return this.http.get<Hamster[]>(`/api/hamsterByName/${name}`)
 }

Operator switchMap() is automatically canceling http request

The last thing that I want to mention here is a simple component that is being used for showing already typed words in search. To achieve this, I’ve used filter() and scan() operators from RxJS. Operator filter() simply filters this (hot observable) searchInputChange$ results - so I can only check the things that make sense. And scan(), on the other hand, accumulates data similarly to reduce(). The only difference here is that “RxJS reduce()” waits until observable completes, whereas scan() takes every value from the pipeline.
 

hamster-search.component.ts

   this.typedWords$ = this.searchInputChange$.pipe(
     filter(f => !!f),
     scan((a, b) => {
       return a.concat(b).slice(-20);
     }, [] as string[]),
     tap(() => this.scrollBottom(this.side)),
   );

 Optimization with debounceTime(), distinctUntilChanged()

Our search component works pretty well, but in a real-life scenario requesting so much data would not be the ideal stuation. We can solve this issue by adding operators debounceTime(), distinctUntilChanged() to our hot observable searchInputChange$. This will prevent our search input from producing too much data and, also, we will not request server with the same query again and again in a row.
 

hamster-search.component.ts

 @ViewChild('search', { static: true }) search?: ElementRef<HTMLButtonElement>;
 searchInputChange$ = defer(() => fromEvent<KeyboardEventType>(this.search?.nativeElement as any, 'keyup'))
   .pipe(
     map(c => c.target.value),
     debounceTime(500),
     distinctUntilChanged(),
     share(),
   );

Optimized version of hamsters search

Applying likes to hamster picture (exhauseMap) - during loading message user can press the “like” button with no effect because of the exhaust map 

RxJS is especially useful when we are dealing with complex user interaction on social media applications. Clicking and unclicking “like” buttons + a whole lot of things that can happen between persisting our post (or comments) can be handled nicely by operator exhaustMap(). 

Feed hamster with love is another simple app with hamsters posts that can be “loved” by the user. The clicking heart part can be based on Subject() which can be also later chained with another observable saveLoveClick$ - this one will be saving the love counter for each hamster. 

Subject classes are the perfect way of creating observables from your own manually triggered events. In this example, I can react to an event from component ‘app-hamster-post’, but also I could trigger the .next() event of loveClicking$ observable from any place inside feed-hamster-with-love.component.ts component class. Beside the Subject class, there is also a very commonly used BehavioralSubject class, which is similar to Subject, but it has a value at the beginning.
 

feed-hamster-with-love.component.ts

loveClicking$ = new Subject<HamsterPost>();

saveLoveClick$ = this.loveClicking$.pipe(
   exhaustMap(hamster => this.api.applyLoveTo(hamster))
 ).subscribe();

feed-hamster-with-love.component.html

<ng-container *ngFor="let hamster of (hamsters$ | async)">
 <app-hamster-post [hamster]="hamster"
                   (loveButton)="loveClicking$.next(hamster)"></app-hamster-post>
</ng-container>

In our search, by using exhaustMap() we prevent the api function applyLoveTo() from being overloaded with ”love increase” requests. In the real world, with a real server, this would save us a lot of cpu power (because each request is taken “seriously” and the user client in the browser waits for each request to be completed, before starting another).

feed-hamster-with-love.service.ts

applyLoveTo(hamster: HamsterPost) {
   return new Observable(obs => {
     hamster.isLoading = true;
     setTimeout(() => {
       if (isNaN(hamster.loveLevel)) {
         hamster.loveLevel = 0;
       }
       if (hamster.loveLevel <= 100) {
         hamster.loveLevel = Number(hamster.loveLevel) + 5;
       }
       hamster.isLoading = false;
       obs.complete();
       this.snackBar.open(`DONE`, void 0, {
         duration: 1000
       });
     }, 3000);
   });
 }

Our hamsters app can be tested by hitting the “love” button many times.  With exhaustMap(), we can only apply love to 1 hamster at a time - other clicks/tries are ignored. The situation would dramatically change with, for instance, mergeMap() operator instead exhauseMap()  - we would see the request race condition.

Final app with hamsters posts

Avoiding memory leaks subscribe/unsubscribe or takeUntil()

Ending this section, I want to mention how to avoid memory leaks using RxJS. Usually, it is a very common situation where we manually subscribe to observables and we forget to unsubscribe from them (just to let you know, in templates async pipe - Angular internal mechanism does “unsubscribe” for us). This mistake can be invisible at first, however, we will likely find out about it later – doing some deeper tests.

Anyway… There are 2 solutions to avoid situation: first one, we can push the subscribe() result into some array (inside component) and later in ngOnDestroy() iterate over this array to unsubscribe from each observable… and in the second one, we can use operators takeUntil(with some observable) that will complete all observables in the Angular’s ngOnDestroy() hook.

Both solutions are valid, but please be careful with the second one. It could happen that when you try to cache/replay values with the shareReplay() operator - takeUntil() may work in a different way to what you expect.
 

search-results-posts-interactions.component.ts

destroy$ = new Subject<void>();

search-results-posts-interactions.component.ts

   this.route.queryParams.pipe(
     takeUntil(this.destroy$),
     tap((params) => {
       if (Object.keys(params).length === 0) {
         this.setTab(this.SEARCH)
       } else {
         const tab = params['tab'] as HamsterRouter;
         this.expanded[tab] = true;
       }
     })

search-results-posts-interactions.component.ts

ngOnDestroy(): void {
   this.destroy$.next();
   this.destroy$.unsubscribe();
 }

What’s next? Closing remarks 

Last but not least, after this tutorial, I would recommend you checking out ngxr - which is a state management engine for Angular. It is a perfect library if you are planning to have complex interactions inside your components - and it can really simplify things. Besides that, I would also recommend you to look at: RxJS unit-tests, marble diagrams, caching/reusing data, http interceptors and operators: take(), toArray() ,timer(), retry(), retryWhen(), delayWhen(), zip(), takeWhile(condition), takeUntil(observable),shareReplay(), pairwise() ,windowsCount(). 

I hope that my article gives you a quick inside into the RxJS/Angular world (or React, or Vue, or whatever you prefer… even pure, vanilla JS! ) and it will encourage you to write more complex things without any fear. Good luck!

Read more: React & Rx.js – 5 Things You Should Know About It