Dangers and Treasures of RxJS
Tags: angular, rxjs, observables, webdev
Published: 2023-03-07 23:25:39

In this article, I'll show you how to avoid known dangerous pitfalls, and how to use some useful tricks, while working with RxJS observables.

## Dangerous Operators 🏴‍☠️

Some operators can cause infinite loops or make your observable behave not as you expect. Let's start with the most dangerous.

### [combineLatestWith](https://rxjs.dev/api/index/function/combineLatestWith)()

It's a very handy operator - usually, you want to pass there one or multiple observables and react to them. 

```ts
// Mocks of the sources, just for illustration
const showNames$: Observable<boolean> = of(false);
const showEmails$: Observable<boolean> = of(false);
const users$ = of([{name: 'a', email: 'a@a'}]);

// Now let's create an observable 
// using `combineLatestWith()`

const usersList = users$.pipe(
  combineLatestWith(
    showNames$,
    showEmails$,
  ),
  // when `showNames$` OR `showEmails$` generate a new value,
  // we'll get latest values from all of them:
  map(([users, showNames, showEmails]) => {
    if (showEmails && showNames) {
      return users;
    }
    return users.map((user) => ({
      ...user,
      name: showNames ? user.name : '',
      email: showEmails ? user.email : ''
    }));
  })
);

```

If at least one of the observables passed to `combineLatestWith()` emits, you'll get the last emitted values from every observable. But this convenience comes with 2 dangerous pitfalls:

**⚠️ #1**: It will not emit until **_every_** observable produced at least one value.  

If you are subscribing to some [hot](https://dev.to/oz/hot-and-cold-observables-4a41) observable, you have no guarantee that you will receive any values at the moment of subscribing.

Every observable passed to `combineLatestWith()`, should be double-checked - you should be sure that at least one value will be emitted.

If you are not sure about that, use [`startWith()`](https://rxjs.dev/api/index/function/startWith) and pass there some default value that your code will expect and can handle. Otherwise, it's too easy to create an "eternally hanging" observable when combineLatestWith() is involved.

**️️⚠️ #2**: If in your code you will trigger one of the observables (passed to `combineLatestWith()`) to emit a new value, you'll create an infinite loop.

Example:

```ts
observable$.pipe(
  combineWithLatest(toggle$),
  map(([value, currentToggleValue]) => {
    if (value && currentToggleValue) {
      toggle$.next(true); // infinite loop
    }
  })
);
```

### [distinctUntilChanged](https://rxjs.dev/api/operators/distinctUntilChanged)()

A very handy operator, what could go wrong?

When you need to detect changes in strings, numbers, and other primitive types, everything is fine. But if you need to detect a change in an object, then this operator might work not as you expect - if the object remains the same, and only some key or sub-key has changed - this operator will not notify you about changes. 

In such cases, you can set a "comparator" argument, or use [`distinctUntilKeyChanged()`](https://rxjs.dev/api/index/function/distinctUntilKeyChanged) operator.

### [forkJoin](https://rxjs.dev/api/index/function/forkJoin)()

You mostly will use it to run multiple requests in parallel and get the responses when all of them are complete.

There are 3 dangerous things in this operator:

**⚠️ #1**: As said in the documentation: "If **_any_** given observable errors at some point, forkJoin will error as well and immediately unsubscribe from the other observables".

In most cases, it's not what you expect, but a workaround is quite easy: add `catchError(() => of(someValue))` to every request: 

```ts
let ob$ = forkJoin([
  req1.pipe(catchError(() => of(someValue))),
  req2.pipe(catchError(() => of(someValue))),
  //...
]);

// or, if you have requests in an array

let ob$ = forkJoin(
  requests.map(req => req.pipe(catchError(() => of(someValue))))
);
```

Although, do not return `EMPTY`, because of the next point:

**⚠️ #2**: As said in the documentation: "…whenever any of the given observables completes without emitting any value, forkJoin will complete at that moment as well and it will not emit anything either, even if it already has some last values from other observables".

It happens not so often, but it still might happen (if some observable in the list returned `EMPTY` or actually didn't complete). And in your app it might create quite a nasty bug - it will look like your app doesn't respond and nothing happens.

To avoid it, you can use [`defaultIfEmpty()`](https://rxjs.dev/api/operators/defaultIfEmpty):

```ts
const ob$ = forkJoin(
  requests.map(req => req.pipe(
    catchError(() => of(someValue)),
    defaultIfEmpty(someValue)
  ))
);
```

**⚠️ #3**: As said in the documentation: "if there is an observable that never completes, forkJoin will never complete either".

It is pretty much possible for HTTP (XHR) requests to hang. In this case, an easy solution is [`timeout()`](https://rxjs.dev/api/index/function/timeout) operator.

For other (non-HTTP, hot) observables there is a chance that you are subscribing to watch an already emitted event. There is no common solution for such cases.

### [takeUntil](https://rxjs.dev/api/index/function/takeUntil)()

The dangers of this operator are easy to avoid: it should be last in the sequence of the operators. With two exceptions: it should be before `shareReplay({refCount: false})` and before aggregating operators ([list](https://github.com/cartant/rxjs-tslint-rules/blob/v4.14.2/source/rules/rxjsNoUnsafeTakeuntilRule.ts#L41-L62)).

It might sound a bit tedious, but you can just [delegate control to the linter](https://github.com/cartant/eslint-plugin-rxjs/blob/main/docs/rules/no-unsafe-takeuntil.md) and forget about it.  
More information you can find in this [article](https://ncjamieson.com/avoiding-takeuntil-leaks/).

### Careless usage of switchMap()

If you read my previous article, you know the difference between `switchMap()`, `concatMap()`, `mergeMap()` and `exhaustMap()`. But, [turns out](https://ncjamieson.com/avoiding-switchmap-related-bugs/), not everyone has read that article, and people carelessly use `switchMap()` in places where it might cause non-desired effects.

It doesn't mean that `switchMap()` is bad or has some flaws - it's a super-useful and good operator, people just misuse it.
Read my previous article and use the correct operators ;)

### Unbound Methods

People are lazy, so we can write code like this:

```ts
ob$.pipe(
  catchError(this.errHandler)
);
```

It will work until `this.errHandler()` does not touch this - because `this` will be [something unexpected](https://ncjamieson.com/avoiding-unbound-methods/).

It's pretty easy to avoid:

```ts
ob$.pipe(
  catchError(() => this.errHandler())
);
```

## Treasures 💎

### [takeUntil](https://rxjs.dev/api/index/function/takeUntil)()

This operator will let you avoid memory leaks in Angular Components  - just add it (last, as noted) to the list of operators before you `subscribe()` and the issue is solved!  
It's the most simple and most convenient solution.

There is a [super-handy linter](https://github.com/cartant/rxjs-tslint-rules#rxjs-prefer-angular-takeuntil) for that.

You do not need this operator if you subscribe using `async` pipe.

### [finalize](https://rxjs.dev/api/operators/finalize)()

When composing the sequence of operators, if you think "after all that ends we should do this", then you might want to use `finalize()`. 

It is much more reliable than just adding some code to `subscribe()` or `map()`, `tap()`, `catchError()` - this operator will be executed even if an observable completes without emitting any values. And in some cases, it's pure gold.

```ts
ob$.pipe(
  tap(() => this.patchState({loading: true})),
  finalize(() => this.patchState({loading: false})),
  exhaustMap(() => this.api.createNewSomething())
);
```

### [first](https://rxjs.dev/api/operators/first)()

Use this operator when you expect that an observable will emit only one value (like an HTTP request), but you can't guarantee it - usually, it happens in functions/methods/effects where you accept some observable as an argument. 

Without this operator, very curious side effects and bugs might appear, if an observable will start emitting multiple values when your code is not ready for that.

### [throttleTime](https://rxjs.dev/api/operators/throttleTime)()
If you ever tried to reproduce this logic: "accept first emitted value, start some side effect, then ignore any consecutive values for N milliseconds, then if there was some new value emitted during this time - emit it too", then `throttleTime()` would save your day. 

It works with similarities to `debounceTime()`, although its `config` argument has parameters `leading` and `trailing`, and it allows to create more sophisticated behaviors.

Example for the logic, mentioned above:

```ts
input$.pipe(
  throttleTime(250, undefined, {leading: true, traling: true}),
  distinctUntilChanged(),
  switchMap((input) => this.api.loadItems(value))
);
```

### [tapResponse](https://ngrx.io/api/component-store/tapResponse)()
And the last in this list will be a third-party operator (not from RxJS).
You can take it from [@ngrx/component-store](https://www.npmjs.com/package/@ngrx/component-store) library.

This operator will help you to don't forget to handle error cases:

```ts
ob$.pipe(
  tapResponse(
    (result) => this.handle(result),
    // the next argument is required, so you will not forget it
    (err) => console?.error(err)
  )
);
```

_If you know any other dangers and treasures of RxJS - share them in the comments!_

---
💙 _If you enjoy my articles, consider [following me on Twitter](https://twitter.com/eugeniyoz), and/or [subscribing](https://medium.com/@eugeniyoz) to receive my new articles by email._  
🎩️ _If you or your company is looking for an Angular consultant, you can [purchase my consultations on Upwork](https://www.upwork.com/services/consultation/development-it-evgeniy-1696285343741362176?ref=project_share)._