RxJS in Angular

First thing that you see as soon as you start learning Angular is RxJS. It looks intimidating and adds up to the mess in the head caused by learning of the framework. There is a reason to that. RxJS by itself offers more features than Angular itself and there are no analogues to it so it’s totally unfamiliar to you. Let’s get it sorted out.

First off, what is RxJS?

It’s an implementation of observer pattern in JavaScript language (the library is available in many other languages). It allows us to do a so called “reactive programming”. Reactive programming means that you build your logic as a response to some events that happen during its lifetime.

Why use RxJS and not Promises/callbacks?

When you need to handle a series of data transformations (e.g. you make a request, based on its response do some transformation and make another request etc.) you need to write a lot of manually leveraged logic. It’s not unified. It’s verbose.

On the counterpart, RxJS enables us to write a more streamlined code, following already established conventions. Moreover, it already has a lot of built-in operators that enable you to write a declarative code without a need to manually implement such functionality as debounce etc.

If you were to use Promises, there is a problem — Promise delivers only one value, whereas in RxJS you work with a stream of events (it’s like your Promise could resolve multiple times). As for the callbacks we all know that it would be hell. Let’s not get there.

RxJS building blocks

Source of events (Observable, Subject) — as the name says it’s the source of events in your application (e.g. a state was changed, response from the server was received etc.).

Consumer (subscriber) — these are your functions that will handle the events (emitted data, errors, completion of a stream).

Operators — these transform the data, create sub-streams. This is the place where you would put some intermediate logic that is not related to the final data handling.

Observable

This is the source of your events. All subscribers will have to subscribe to an observable in order to have it working. With Promises it’s enough to create it and the internal logic is already being executed. However, with Observables the initializing logic is not executed until you subscribe to the observable. This is vital to remember.

Similar to promises you can either create your own observable that emits some data, or use a predefined function from the library which will cover most common scenarios.

Let’s see how we can create an observable.

First, we create an observable. The constructor accepts a callback function which accepts a Subscriber object. This object has 3 methods that we are interested in:

• next — this is similar to Promise’s resolve, however it can be an unlimited amount of times as long as the observable is not complete or in error state.
• error — this is similar to Promise’s reject. This will prevent your observable from emitting any further events.
• complete — this tells your subscriber that your Observable won’t emit any values. Only after that will JavaScript’s garbage collector able to clean the memory from it. So it’s crucial to always end your observable when you don’t need it anymore. Otherwise, you’ll end up with memory leaks and perhaps some callbacks still working despite your expectations.

Subscriber

It’s a handler for your observable’s events emitted. It is passed into the subscribe method either as 3 functions (next, error, complete) or as on object with 3 properties (next, error, complete).

• next — is responsible for values emitted by the observable. It’s like Promise’s then callback functions.
• error — is responsible for handling an error event.
• complete — is called at the observable’s completion.

Arguments are optional to the subscribe method and you can simply do source.subscribe(), but only if you really don’t care of the results.

If you were to make a request in Angular you would write

this.httpClient.get(`http://my.url.com/users`).subscribe(
  (users) => {
    this.users = users
  },
  (error) => {
    console.log(`cannot fetch users`, error)
  }
);

Here, we pass 2 callbacks: the first one is to handle the returned users data from the request; the second one will be called if there is a network error to happen.

Since it’s an HTTP request it only emits 1 value and then automatically completes, so there is no need to worry about memory leaks here.

Keep in mind that by default observables are single-cast. This means that if we store an observable into a constant and then subscribe twice to it — the code will be executed twice (for HttpClient you would make 2 separate requests). It’s like as if you were doing new Promise for each subscription.

Also there are multi-cast observables — this means that 1 source observable can have any amount of subscribers and it won’t be executing its logic for each subscription over and over again. For example fromEvent is a multi-cast observable. It allows you to subscribe to a document’s event (e.g. click ). Since the source of the events is outside of the observable (clicks are not generated when you do new Observable , but rather when you click with your mouse) it’s reasonable that observable will execute only once and then when clicks happen it will just tell about them to each subscriber it has.

Operators

Operators are the most interesting and most diverse part of the RxJS. They enable us to do data transformations, delay emits etc. Under the hood they are just functions which return a new observable (it subscribes to the source observable, does some logic and then emits values in the new returned observable), so no magic there. Operators can be chained and are passed as arguments to the pipe method of observable.

source.pipe( // emits { name: `test`, age: 12 }
  take(1), // Takes 1 value, then unsubscribes
  pluck(`name`), // Plucks a property of the object. Will return `test`
  map(v => `value ${v}`) // Allows us to do transformations over the values
).subscribe(data => {
  console.log(data) // Will log `value test`
});

Let’s have a look at a few most used ones in Angular:

• take — specifies the amount of emissions subscription will take before unsubscribing. E.g. if you listen for a click event on a button in the modal you want to unsubscribe after the first click, therefore you specify “1” and it will unsubscribe for you.
• pluck — enables easy and declarative way to extract a property from an object. It was brought to my attention, that this is no longer the case since RxJS v8. map should be used instead. However, you will still find this operator in the existing codebases.
• map — similar to Array.prototype.map, but instead of instantaneous iteration over an array it will iterate over each value that goes from the source and immediately pass through the result.
• switchMap — allows us to return an observable from the callback function, this way we can do requests based on a data that was received previously.

this.httpClient.get(`http://api.example.com/users/1`).pipe(
  pluck(`id`), // get user id from the request
  switchMap(id => this.httpClient.get(`http://api.example.com/posts?userId=${id}`)) // Here we return a new observable
).subscribe(/* This will receive the value from the posts request */)

• takeUntil — allows us to unsubscribe when an observable passed to takeUntil emits a value. This is very convenient as we can unsubscribe all the subscribers in our component when it gets destroyed.

@Component(…)
export class MyComponent {
  unsubscribe$: Observable<void>;

  ngOnInit(): void {
    this.unsubscribe$ = new Subject(); // We create an observable that we will use as a notification that component gets destroyed

    fromEvent(document, `click`).pipe( // Here we subscribe to some observable that will still be emitting values after the component gets destroyed
      takeUntil(this.unsubscribe$) // We use `takeUntil` operator and give it our notifying observable
    ).subscribe(/* Handle clicks on the document */)
  }

  ngOnDestroy(): void {
    this.unsubscribe$.next(); // Emit event so `takeUntil` will unsubscribe from the source observable
    this.unsubscribe$.complete(); // Complete the observable, so there will be no leftovers and garbage collector can remove it from the memory
  }
}

Subjects

Subjects are observables which don’t have their initializing logic and API is public. Let’s compare how we would work with them

const source1 = new Observable(subscriber => {
  // Here goes the initializing logic
  // Subscriber’s API is scoped to this callback, so you can’t call it from the outside

  subscriber.next(`test`);

  // The logic whill execute for each subscriber
});

const source2 = new Subject(); // Subject doesn’t accept any arguments
source2.next(`value`); // Subscriber’s API is available on the returned object and therefore values can be emitted by anyone having access to the object
// Subscriber’s logic won’t execute on subcribtion.

source2.subscribe(value => console.log(value)); // It won’t log `value` because it was emitted before we subscribed
source2.next(`value2`); // This value will be logged as it’s being emitted after the subscription happened

So, on the example above we can already see a clear difference. Subjects are independent. They are multicast observables by default. They don’t have any initializing logic. Their API is public and can be accessed on the object.

It’s convenient from one side, but from the other if you expose this Subject from your service as a public property — any part of the application will be able to emit a value, which is not always a desired behavior. You can fix it by using asObservable method of the Subject instance. Returned value will not contain any methods that can be used to emit values in the observable. It can be only subscribed to.

Conclusion

RxJS is a very powerful library with a vast features set. It’s really worth to learn the library as it will make your life much easier. Lots of the logic you would write manually, can be easily written with existing RxJS operators in a declarative way which is easy to read and understand.