Made withAngularTutorial

Learn RxJS Angular, RxJS pipe, async pipe, RxJS observables, Behaviorsubject, Combinelatest

M

Monsterlessons Academy

Creator on the Angular channel

Watch on YouTube ↗

118K viewsJun 28, 202219:17 runtime

Master essential RxJS concepts for building robust Angular applications. This guide focuses on practical, everyday techniques, starting with creating observables from data using `of`. Learn why the `async` pipe is superior to manual subscriptions for cleaner code and automatic…

Chapters

#angular#tutorial#frontend

Core Concepts

RxJS allows you to work with asynchronous data as streams (Observables).
A dollar sign suffix ($) is a common convention for naming variables that hold an Observable (e.g., users$).
Streams are lazy; they don't emit values until they are subscribed to.

Creating and Rendering Streams

of(): Creates an Observable stream that emits a static value (like an array or object) you provide, and then completes.
```
users$ = of([ { id: 1, name: 'John', isActive: true } ]);
```
Async Pipe (| async): The preferred way to subscribe to Observables in Angular templates.
- It automatically subscribes when the component is created and unsubscribes when it is destroyed, preventing memory leaks.
- It unwraps the value from the Observable for direct use in the template.

Important
Always prefer the async pipe over manual .subscribe() calls. Manual subscriptions require implementing OnDestroy to unsubscribe, which adds boilerplate and creates a risk of memory leaks.

Transforming Streams

The pipe() method is used to chain together RxJS operators to create a new Observable based on the source.

map(): Transforms each value emitted by the source Observable. It's used for synchronously changing the shape or type of data.

// From an array of user objects to an array of names
userNames$ = this.users$.pipe(
  map(users => users.map(user => user.name))
);

filter(): Emits the value from the source Observable only if it passes a specified condition. Note that it filters the entire emission, not individual items within an emitted array.
```
// Only emits the 'users' array if every user in it is active
activeUsers$ = this.users$.pipe(
  filter(users => users.every(user => user.isActive))
);
```

Managing Component State

BehaviorSubject: A special type of Observable that holds a "current value."
- It requires an initial value upon creation.
- You can push a new value into the stream at any time using the .next() method.
- This is ideal for managing dynamic state, like data fetched from an API.
```
// Initialize with null, representing data that hasn't loaded yet
user$ = new BehaviorSubject<{ id: string; name: string } | null>(null);

// Later (e.g., after an API call), update it with fetched data
this.user$.next({ id: '1', name: 'John' });
```

Handling Events and Combining Streams

fromEvent(): Creates an Observable from a DOM event, allowing you to treat user interactions as streams.
```
// Create a stream of click events on the document
clicks$ = fromEvent(document, 'click');
```
combineLatest(): Combines multiple Observables. It emits an array of the latest values from each source stream whenever any of the source streams emit a new value.

Tip
A powerful pattern is to use combineLatest to merge all component streams into a single data$ object. This dramatically simplifies the template by requiring only one async pipe.

data$ = combineLatest([
  this.users$,
  this.userNames$
]).pipe(
  map(([users, userNames]) => {
    // Map the array output to a structured object
    return { users, userNames };
  })
);

Then in the template:

<ng-container *ngIf="data$ | async as data">
  <!-- Now you can easily access data.users, data.userNames, etc. -->
  <div *ngFor="let user of data.users">{{ user.name }}</div>
</ng-container>

Introduction to Practical RxJS in Angular

In this video I want to talk about RxJS, RxJS streams, observables, and how to use RxJS inside Angular. And actually, inside the RxJS we have hundreds of different methods and it is really difficult for beginners to understand all these methods together with Angular. This is why in this video I want to focus only on things that I am using inside RxJS every day because they are super efficient and is needed in every single project.

Creating Observables from Data with `of()`

And the first thing that I want to talk about in RxJS is the function of. As you can see here, I have an array of users, and this is just a plain JavaScript array. And let's say that we want to create a stream from some data. And for this, we can create users$ and with dollar, we mark our streams. And here we can write of, and this is a function of RxJS, and inside we can pass whatever we want. This is just a plain data. In our case, we can write directly this.users and we must import this of from RxJS. So what a function does, it converts plain data in the stream. And now as you can see here, users$ is an observable of an array with objects of ID, name, and isActive. And this is really a nice way to just convert your plain data inside a stream if you for some reason want to work with them like with a stream.

Using the `async` Pipe vs. Manual Subscriptions

And typically, what all people are doing when they're just starting with RxJS or Angular, they simply write here implements OnInit and after this they just subscribe to all the streams. We subscribe and use plain data inside. So they are writing something like this, users.subscribe and we are getting here our list of users. And now we are doing something with them for example console.log and here we are getting users. This is totally fine as a beginner approach, as you can see it is working. We are getting here inside subscribe access to our array of users but actually this is not how it is intended to be used because RxJS inside Angular shines when we are using it together with async pipe. And actually, when we are using subscribe as little as possible. So the typical use case here will be to render this array of the users. And for this, we should not use subscribe. We can use async pipe inside Angular. This is why what we can do here, we can just write here a div and we want to loop through our users. This is why here we are just writing ngFor and we want to access every single user inside of a stream of users. But this is a stream, we can't write code like this. After this, we must write async pipe. And in this case Angular will itself subscribe to the stream and when it gets data, it will convert these users to just plain array of users. And here inside this div now we have access to every single user and we can simply render here user.name for example. Let's check this out. As you can see here, we just see our plain array. We don't need to bother here with subscriptions or with streams. Which actually means if you can use async pipe inside Angular instead of subscribe, just do it. It is much better because in this case, Angular itself creates this subscription and destroys it when you are destroying your component. If you are doing subscribe like this, then you must yourself unsubscribe inside onDestroy. If you won't do it, then you will have a hanging subscription.

Transforming Streams with `pipe` and `map`

But typically people don't understand how they can transform data with RxJS. This is why they just want to get rid of RxJS and just write subscribe and inside plain JavaScript code. This is a bad approach. This is why here I want to show you some methods that will help you to work with subscriptions. Let's say here we have an array of users and this is actually a stream.

And we can say that these users maybe are not available inside this component so we simply work with this stream. These users might come for example from an API. So this is just a stream and we want to get here an array of our user names, which actually means here from every single object we want just to take a name. And typically people will write subscribe and write it to some property but we can do much better. We can create here a new stream which is called usernames and here we want to take our stream, this.users$, and now we can transform it. And to transform a stream we must use a pipe. This is mandatory. But as pipe, this is a function which helps us to organize one by one all our functions inside, which actually means here inside we have function one, function two, and function three for example. Which actually means this is a list of functions which will transform one by one our stream. So these users are going through function one, then function two, then function three, and the result of that function will be written here inside usernames. So how we can use it?

Here for example, inside our pipe, we can write another function which is a map. And actually map is the most popular function inside of RxJS and we use map if we want to transform our data, which actually means here inside map now, we're getting access to our users and we can simply just return these users and don't do anything with them. As you can see here, user is an array of our objects, exactly what was here in the stream. And the main idea is that here inside map we can write synchronous code like plain JavaScript, which actually means if here we want to get our names, we can simply write users.map, we're getting access to every single user, and here we can write user.name for example, and we will map every single user. Which actually means here in usernames$ as you can see here we get an observable of string array. So it is not object array anymore, this is an array of names. And actually for this, we don't need subscribe, we don't need to create additional local properties and check that we fulfilled our subscribe. Here we simply have a stream which is based on our first stream. And now we can just jump inside our HTML, copy paste this code with ngFor and write it for our usernames$ with the async pipe. And here we are getting access to every single username and we can simply render it. Here it is username. And now I just want to comment out our first div, we don't need it anymore. And as you can see in browser it is still working just as intended but here we are using completely another stream which we just created based on our first stream. So if you need to transform data inside the stream to another stream it makes a lot of sense to leverage map function inside RxJS.

Filtering Streams with the `filter` Operator

Another super important function is filter inside RxJS. And actually you can understand from the name, it filters our data, but actually it is not working like filter inside JavaScript. When we are filtering an array inside JavaScript, we simply leave less elements, but when we are filtering a stream it means that we're just coming through the stream later if the predicate is true. What does it mean? For example, here we can create filteredUsers and this is also a stream. And we want to get here only users when they are active, which actually means if every single user in our array of users is active, then we will get here our users. If one of the users is inactive, then we will never get here, this stream will never be fulfilled. How we can write it? We can write here this.users$ and we want to use also pipe as always, and inside we pack not map, but we are packing here filter. And inside we are getting access to our users and we must return here true or false, which means if we are returning here true as a predicate, then the stream will be directly fulfilled. But here we want to write some logic and we want to check that every single element inside users is active. This is why here we want to check user.isActive. So a function inside filter must return true or false. In our case here we are checking every single element for isActive.

And now here, let's try to render our filtered users. For this, I will copy paste the first div that we have and just uncomment it here. And we want to go through our stream filteredUsers. Here we have access to every single user and we are rendering it. Let's check this out. As you can see here is our list. And I want to comment out this div because we don't need it anymore. As you can see here, all our names are there which actually means the stream was fulfilled. But what will happen if one single user here will be false? In this case here inside filter we will get false and this stream won't be filled with data. And as you can see here, nothing was rendered. And actually if you will try here to uncomment this user subscribe and change it to filtered users, then you will never get inside subscribe. As you can see here, it is not happening because it is filtered, our user is inactive. But here now we can change it to true, and at this very moment we are getting here users inside console and our users are rendered here. And it makes a lot of sense to write code like this when you want to filter your stream. And as I already said, we are using pipe and actually we can combine different things. For example, here we could take our map completely from here from our pipe and just put it here after our filter, which actually means here is our first function inside pipe, here is our second function to map our data. This is completely fine and typically we combine our functions like this.

Managing Dynamic State with `BehaviorSubject`

Another thing that you will use every single day is BehaviorSubject, and it might sound really scary but it is not. What we can do here, we can create a new stream which is a user and here we simply write new BehaviorSubject. And inside we must pass what data we have inside. So for example, I would say that we have here id string and name string, or we have here null. Which actually means inside this stream we have two possible values, this object or a null. And here we must open round brackets and set the default value. And typically we'll write here null, which is a default value. While we're writing code like this, typically we want to fetch some data from the backend, from the API, and this will be your stream which you want to fill with data. Which actually means you have a page where you are rendering your user and by default you don't have any user. This is why inside stream you have a null, but then after some time, you fetched the data and you put them inside your user stream. At that very moment, your stream must be automatically rendered. This is exactly why we need BehaviorSubject. BehaviorSubject is a stream that we can update. And how we can do that? For example here inside ngOnInit what we can write is setTimeout to show that this is an API call. And here inside setTimeout, for example with two seconds, I want to update this user. For this we are writing this.user$ and here .next. And we are using next function when we want to update our behavior subject. And as you can see here, we can't just throw foo, it is invalid. We must write here either null or our valid user. This is why here I will write id: '1' and name: 'John', and this is the valid type of data. So once again, what we have here, we have a Behavior subject which is just a stream and by default it is null. Then at any moment we can use this user.next and set some other data there. And this will update our stream, which actually means in different places we can subscribe to the stream and then all these places will be notified about new data. This is why here we can write this.user$.subscribe and here we're getting access to our user. And now here I just want to console.log(user) so you can understand how it is working. I'm reloading the page. As you can see here user is null and then after two seconds we are getting user and here is our object. Which actually means this is a typical approach when you are working with API. By default your data is null, then after you fetch them you can simply write this user.next for example and then you are setting data inside your stream and your stream will be rendered on the page.

Cleanly Rendering Data with `ngIf` and `async`

This is why how we can render this data. Now we can just write here div and we want to write here ngIf because actually we want to render this user only when it is not null. For example, we want to render a name and for this we can write here user$ | async pipe. And actually a lot of people are writing like this. Inside this div they simply render user$ then async and actually this can return you null. This is why they are writing ?. and then name, which actually means here we are waiting for this async pipe. If it is there and it is not null then we are rendering name. As you can see this code is working, here is our John, but this code is really bad. Why is that? Because here we first of all create two different subscriptions and yes, Angular creates them for us, but it doesn't matter. This is first subscription, this is second subscription. It doesn't make any sense and also this code is not looking really good. What we can write instead, we can write here as user. In this case here we're creating inside this div a local property user. And now inside we can write user.name. And actually this code is working completely fine because first of all this ngIf checks if our user is there, which actually means if our user stream is null then we won't come here inside div and we don't need to write any checks like with question mark that this name property exists. We will get here only when our user will be fulfilled.

Handling DOM Events as Streams with `fromEvent`

Another important function that you will use from time to time is fromEvent inside RxJS. And actually fromEvent helps you to work with DOM events like with streams, and this is extremely efficient. For example, here I can write documentClick and this is a stream, and here I can use fromEvent. And inside I must first of all pass a DOM element, this is a document, and then the event, for example click. In this case here, I am not working with DOM events. I have here a stream and I can work with a stream just like with any stream. For example, here we can write this.documentClick.subscribe and here we are getting our event and here we can just write console.log(event). Let's check this out. I'm reloading the page and I'm just clicking on our page and as you can see here we're getting our event just like normal subscription to the document click. But here you might say, okay, but it doesn't make any sense for me. Using this fromEvent is confusing. I can just write plain JavaScript with DOM. And yes, you can. But the main idea of streams is to combine them, which actually means if you have just a single event, it doesn't make a lot of sense. But if your API calls are streams, if your data for your state are also streams, and your DOM events are streams, then you can combine them together and this is really easy inside RxJS. This is why RxJS by default is not better than a single event on the document. But if you want to write more advanced stuff inside Angular it makes a lot of sense to use RxJS everywhere.

Combining Multiple Streams with `combineLatest`

And the last method that I want to show you is extremely important and this is exactly a topic of combining our streams of data. And what I want to show you is a function which is called combineLatest, and we're using it to combine our streams. And typically what you want to do, you want to take all your streams for the component and combine them in one single object. In this case, it is much easier to support a component and your markup inside HTML will be much cleaner. Let's try this out.

Now, as you can see here, we have our users$, usernames$, and filteredUsers$. And let's say that we need all these data, all the streams inside our HTML. And what we can do, we can combine them in a single stream of data. This is why here I want to create data$ and they typically name combined streams like data$. And here we can simply write combineLatest. And this is also a function from RxJS, and here inside we're throwing an array of streams. And here first of all, we want to write this.users$, then this.usernames$, and then this.filteredUsers$. And actually here we could also write document click but it doesn't make a lot of sense. The main idea is that any streams can go here inside. But here is a problem. What we will get back? An observable of an array, and inside array we have these three elements. It is not comfortable to work like this inside HTML. This is why what I like to do is map this data. So here we can write pipe, map like we did previously. And here we have access to all three data inside our array. So this is first of all users, then our usernames, and then our filteredUsers. And what we can do here, we can just return this data like an object, and inside we will have first of all our users, then our usernames, and then our filtered users. Which actually means we simply combine our streams, we map them to the object, and now this data$ is an observable of the object. And here we have first of all key users, then usernames, and then filteredUsers. And with this structure it is much easier to write HTML.

Building a Clean UI with a Single Data Stream

We can jump to our app component and now just write here div and what we want to do here, we want to resolve this data stream. This is why here, data$ | async as data. And now inside this div, we have access to everything inside data: data.users, data.filteredUsers, data.usernames. Which actually means we can copy paste all this code inside our div and change it. And now we just work with the data without any streams. This is why here, let user of data.users. Now here we have user names and actually it will be data.usernames and we can remove the async pipe. And here is our filtered users, this is data.filteredUsers. And now just to bring some understanding I want to put dashes here inside markup. Let's check this out. As you can see now on the page it is working and we have three sections. First of all our users, then our usernames, and then our filtered users. And actually this is the best possible approach that you can use if you want to combine different streams of data and render them inside your component. And actually, if you are interested to know how Angular animations are working, make sure to check this video also.