Catch error rxjs

Error handling is an essential part of RxJs, as we will need it in just about any reactive program that we write.

Error handling in RxJS is likely not as well understood as other parts of the library, but it’s actually quite simple to understand if we focus on understanding first the Observable contract in general.

In this post, we are going to provide a complete guide containing the most common error handling strategies that you will need in order to cover most practical scenarios, starting with the basics (the Observable contract).

Table Of Contents

In this post, we will cover the following topics:

• The Observable contract and Error Handling
• RxJs subscribe and error callbacks
• The catchError Operator
• The Catch and Replace Strategy
• throwError and the Catch and Rethrow Strategy
• Using catchError multiple times in an Observable chain
• The finalize Operator
• The Retry Strategy
• Then retryWhen Operator
• Creating a Notification Observable
• Immediate Retry Strategy
• Delayed Retry Strategy
• The delayWhen Operator
• The timer Observable creation function
• Running Github repository (with code samples)
• Conclusions

So without further ado, let’s get started with our RxJs Error Handling deep dive!

The Observable Contract and Error Handling

In order to understand error handling in RxJs, we need to first understand that any given stream can only error out once. This is defined by the Observable contract, which says that a stream can emit zero or more values.

The contract works that way because that is just how all the streams that we observe in our runtime work in practice. Network requests can fail, for example.

A stream can also complete, which means that:

• the stream has ended its lifecycle without any error
• after completion, the stream will not emit any further values

As an alternative to completion, a stream can also error out, which means that:

• the stream has ended its lifecycle with an error
• after the error is thrown, the stream will not emit any other values

Notice that completion or error are mutually exclusive:

• if the stream completes, it cannot error out afterwards
• if the streams errors out, it cannot complete afterwards

Notice also that there is no obligation for the stream to complete or error out, those two possibilities are optional. But only one of those two can occur, not both.

This means that when one particular stream errors out, we cannot use it anymore, according to the Observable contract. You must be thinking at this point, how can we recover from an error then?

RxJs subscribe and error callbacks

To see the RxJs error handling behavior in action, let’s create a stream and subscribe to it. Let’s remember that the subscribe call takes three optional arguments:

• a success handler function, which is called each time that the stream emits a value
• an error handler function, that gets called only if an error occurs. This handler receives the error itself
• a completion handler function, that gets called only if the stream completes

Completion Behavior Example

If the stream does not error out, then this is what we would see in the console:

HTTP response {payload: Array(9)}
HTTP request completed.

As we can see, this HTTP stream emits only one value, and then it completes, which means that no errors occurred.

But what happens if the stream throws an error instead? In that case, we will see the following in the console instead:

As we can see, the stream emitted no value and it immediately errored out. After the error, no completion occurred.

Limitations of the subscribe error handler

Handling errors using the subscribe call is sometimes all that we need, but this error handling approach is limited. Using this approach, we cannot, for example, recover from the error or emit an alternative fallback value that replaces the value that we were expecting from the backend.

Let’s then learn a few operators that will allow us to implement some more advanced error handling strategies.

The catchError Operator

In synchronous programming, we have the option to wrap a block of code in a try clause, catch any error that it might throw with a catch block and then handle the error.

Here is what the synchronous catch syntax looks like:

This mechanism is very powerful because we can handle in one place any error that happens inside the try/catch block.

The problem is, in Javascript many operations are asynchronous, and an HTTP call is one such example where things happen asynchronously.

RxJs provides us with something close to this functionality, via the RxJs catchError Operator.

How does catchError work?

As usual and like with any RxJs Operator, catchError is simply a function that takes in an input Observable, and outputs an Output Observable.

With each call to catchError, we need to pass it a function which we will call the error handling function.

The catchError operator takes as input an Observable that might error out, and starts emitting the values of the input Observable in its output Observable.

If no error occurs, the output Observable produced by catchError works exactly the same way as the input Observable.

What happens when an error is thrown?

However, if an error occurs, then the catchError logic is going to kick in. The catchError operator is going to take the error and pass it to the error handling function.

That function is expected to return an Observable which is going to be a replacement Observable for the stream that just errored out.

Let’s remember that the input stream of catchError has errored out, so according to the Observable contract we cannot use it anymore.

This replacement Observable is then going to be subscribed to and its values are going to be used in place of the errored out input Observable.

The Catch and Replace Strategy

Let’s give an example of how catchError can be used to provide a replacement Observable that emits fallback values:

Let’s break down the implementation of the catch and replace strategy:

• we are passing to the catchError operator a function, which is the error handling function
• the error handling function is not called immediately, and in general, it’s usually not called
• only when an error occurs in the input Observable of catchError, will the error handling function be called
• if an error happens in the input stream, this function is then returning an Observable built using the of([]) function
• the of() function builds an Observable that emits only one value ([]) and then it completes
• the error handling function returns the recovery Observable (of([])), that gets subscribed to by the catchError operator
• the values of the recovery Observable are then emitted as replacement values in the output Observable returned by catchError

As the end result, the http$ Observable will not error out anymore! Here is the result that we get in the console:

HTTP response []
HTTP request completed.

As we can see, the error handling callback in subscribe() is not invoked anymore. Instead, here is what happens:

• the empty array value [] is emitted
• the http$ Observable is then completed

As we can see, the replacement Observable was used to provide a default fallback value ([]) to the subscribers of http$, despite the fact that the original Observable did error out.

Notice that we could have also added some local error handling, before returning the replacement Observable!

And this covers the Catch and Replace Strategy, now let’s see how we can also use catchError to rethrow the error, instead of providing fallback values.

The Catch and Rethrow Strategy

Let’s start by noticing that the replacement Observable provided via catchError can itself also error out, just like any other Observable.

And if that happens, the error will be propagated to the subscribers of the output Observable of catchError.

This error propagation behavior gives us a mechanism to rethrow the error caught by catchError, after handling the error locally. We can do so in the following way:

Catch and Rethrow breakdown

Let’s break down step-by-step the implementation of the Catch and Rethrow Strategy:

• just like before, we are catching the error, and returning a replacement Observable
• but this time around, instead of providing a replacement output value like [], we are now handling the error locally in the catchError function
• in this case, we are simply logging the error to the console, but we could instead add any local error handling logic that we want, such as for example showing an error message to the user
• We are then returning a replacement Observable that this time was created using throwError
• throwError creates an Observable that never emits any value. Instead, it errors out immediately using the same error caught by catchError
• this means that the output Observable of catchError will also error out with the exact same error thrown by the input of catchError
• this means that we have managed to successfully rethrow the error initially thrown by the input Observable of catchError to its output Observable
• the error can now be further handled by the rest of the Observable chain, if needed

If we now run the code above, here is the result that we get in the console:

As we can see, the same error was logged both in the catchError block and in the subscription error handler function, as expected.

Using catchError multiple times in an Observable chain

Notice that we can use catchError multiple times at different points in the Observable chain if needed, and adopt different error strategies at each point in the chain.

We can, for example, catch an error up in the Observable chain, handle it locally and rethrow it, and then further down in the Observable chain we can catch the same error again and this time provide a fallback value (instead of rethrowing):

If we run the code above, this is the output that we get in the console:

As we can see, the error was indeed rethrown initially, but it never reached the subscribe error handler function. Instead, the fallback [] value was emitted, as expected.

The Finalize Operator

Besides a catch block for handling errors, the synchronous Javascript syntax also provides a finally block that can be used to run code that we always want executed.

The finally block is typically used for releasing expensive resources, such as for example closing down network connections or releasing memory.

Unlike the code in the catch block, the code in the finally block will get executed independently if an error is thrown or not:

RxJs provides us with an operator that has a similar behavior to the finally functionality, called the finalize Operator.

Note: we cannot call it the finally operator instead, as finally is a reserved keyword in Javascript

Finalize Operator Example

Just like the catchError operator, we can add multiple finalize calls at different places in the Observable chain if needed, in order to make sure that the multiple resources are correctly released:

Let’s now run this code, and see how the multiple finalize blocks are being executed:

Notice that the last finalize block is executed after the subscribe value handler and completion handler functions.

The Retry Strategy

As an alternative to rethrowing the error or providing fallback values, we can also simply retry to subscribe to the errored out Observable.

Let’s remember, once the stream errors out we cannot recover it, but nothing prevents us from subscribing again to the Observable from which the stream was derived from, and create another stream.

Here is how this works:

• we are going to take the input Observable, and subscribe to it, which creates a new stream
• if that stream does not error out, we are going to let its values show up in the output
• but if the stream does error out, we are then going to subscribe again to the input Observable, and create a brand new stream

When to retry?

The big question here is, when are we going to subscribe again to the input Observable, and retry to execute the input stream?

• are we going to retry that immediately?
• are we going to wait for a small delay, hoping that the problem is solved and then try again?
• are we going to retry only a limited amount of times, and then error out the output stream?

In order to answer these questions, we are going to need a second auxiliary Observable, which we are going to call the Notifier Observable. It’s the Notifier
Observable that is going to determine when the retry attempt occurs.

The Notifier Observable is going to be used by the retryWhen Operator, which is the heart of the Retry Strategy.

RxJs retryWhen Operator Marble Diagram

To understand how the retryWhen Observable works, let’s have a look at its marble diagram:

Notice that the Observable that is being re-tried is the 1-2 Observable in the second line from the top, and not the Observable in the first line.

The Observable on the first line with values r-r is the Notification Observable, that is going to determine when a retry attempt should occur.

Breaking down how retryWhen works

Let’s break down what is going in this diagram:

• The Observable 1-2 gets subscribed to, and its values are reflected immediately in the output Observable returned by retryWhen
• even after the Observable 1-2 is completed, it can still be re-tried
• the notification Observable then emits a value r, way after the Observable 1-2 has completed
• The value emitted by the notification Observable (in this case r) could be anything
• what matters is the moment when the value r got emitted, because that is what is going to trigger the 1-2 Observable to be retried
• the Observable 1-2 gets subscribed to again by retryWhen, and its values are again reflected in the output Observable of retryWhen
• The notification Observable is then going to emit again another r value, and the same thing occurs: the values of a newly subscribed 1-2 stream are going to start to get reflected in the output of retryWhen
• but then, the notification Observable eventually completes
• at that moment, the ongoing retry attempt of the 1-2 Observable is completed early as well, meaning that only the value 1 got emitted, but not 2

As we can see, retryWhen simply retries the input Observable each time that the Notification Observable emits a value!

Now that we understand how retryWhen works, let’s see how we can create a Notification Observable.

Creating a Notification Observable

We need to create the Notification Observable directly in the function passed to the retryWhen operator. This function takes as input argument an Errors Observable, that emits as values the errors of the input Observable.

So by subscribing to this Errors Observable, we know exactly when an error occurs. Let’s now see how we could implement an immediate retry strategy using the Errors Observable.

Immediate Retry Strategy

In order to retry the failed observable immediately after the error occurs, all we have to do is return the Errors Observable without any further changes.

In this case, we are just piping the tap operator for logging purposes, so the Errors Observable remains unchanged:

Let’s remember, the Observable that we are returning from the retryWhen function call is the Notification Observable!

The value that it emits is not important, it’s only important when the value gets emitted because that is what is going to trigger a retry attempt.

Immediate Retry Console Output

If we now execute this program, we are going to find the following output in the console:

As we can see, the HTTP request failed initially, but then a retry was attempted and the second time the request went through successfully.

Let’s now have a look at the delay between the two attempts, by inspecting the network log:

As we can see, the second attempt was issued immediately after the error occurred, as expected.

Delayed Retry Strategy

Let’s now implement an alternative error recovery strategy, where we wait for example for 2 seconds after the error occurs, before retrying.

This strategy is useful for trying to recover from certain errors such as for example failed network requests caused by high server traffic.

In those cases where the error is intermittent, we can simply retry the same request after a short delay, and the request might go through the second time without any problem.

The timer Observable creation function

To implement the Delayed Retry Strategy, we will need to create a Notification Observable whose values are emitted two seconds after each error occurrence.

Let’s then try to create a Notification Observable by using the timer creation function. This timer function is going to take a couple of arguments:

• an initial delay, before which no values will be emitted
• a periodic interval, in case we want to emit new values periodically

Let’s then have a look at the marble diagram for the timer function:

As we can see, the first value 0 will be emitted only after 3 seconds, and then we have a new value each second.

Notice that the second argument is optional, meaning that if we leave it out our Observable is going to emit only one value (0) after 3 seconds and then complete.

This Observable looks like its a good start for being able to delay our retry attempts, so let’s see how we can combine it with the retryWhen and delayWhen operators.

The delayWhen Operator

One important thing to bear in mind about the retryWhen Operator, is that the function that defines the Notification Observable is only called once.

So we only get one chance to define our Notification Observable, that signals when the retry attempts should be done.

We are going to define the Notification Observable by taking the Errors Observable and applying it the delayWhen Operator.

Imagine that in this marble diagram, the source Observable a-b-c is the Errors Observable, that is emitting failed HTTP errors over time:

delayWhen Operator breakdown

Let’s follow the diagram, and learn how the delayWhen Operator works:

• each value in the input Errors Observable is going to be delayed before showing up in the output Observable
• the delay per each value can be different, and is going to be created in a completely flexible way
• in order to determine the delay, we are going to call the function passed to delayWhen (called the duration selector function) per each value of the input Errors Observable
• that function is going to emit an Observable that is going to determine when the delay of each input value has elapsed
• each of the values a-b-c has its own duration selector Observable, that will eventually emit one value (that could be anything) and then complete
• when each of these duration selector Observables emits values, then the corresponding input value a-b-c is going to show up in the output of delayWhen
• notice that the value b shows up in the output after the value c, this is normal
• this is because the b duration selector Observable (the third horizontal line from the top) only emitted its value after the duration selector Observable of c, and that explains why c shows up in the output before b

Delayed Retry Strategy implementation

Let’s now put all this together and see how we can retry consecutively a failing HTTP request 2 seconds after each error occurs:

Let’s break down what is going on here:

• let’s remember that the function passed to retryWhen is only going to be called once
• we are returning in that function an Observable that will emit values whenever a retry is needed
• each time that there is an error, the delayWhen operator is going to create a duration selector Observable, by calling the timer function
• this duration selector Observable is going to emit the value 0 after 2 seconds, and then complete
• once that happens, the delayWhen Observable knows that the delay of a given input error has elapsed
• only once that delay elapses (2 seconds after the error occurred), the error shows up in the output of the notification Observable
• once a value gets emitted in the notification Observable, the retryWhen operator will then and only then execute a retry attempt

Retry Strategy Console Output

Let’s now see what this looks like in the console! Here is an example of an HTTP request that was retried 5 times, as the first 4 times were in error:

And here is the network log for the same retry sequence:

As we can see, the retries only happened 2 seconds after the error occurred, as expected!

And with this, we have completed our guided tour of some of the most commonly used RxJs error handling strategies available, let’s now wrap things up and provide some running sample code.

Running Github repository (with code samples)

In order to try these multiple error handling strategies, it’s important to have a working playground where you can try handling failing HTTP requests.

This playground contains a small running application with a backend that can be used to simulate HTTP errors either randomly or systematically. Here is what the application looks like:

Conclusions

As we have seen, understanding RxJs error handling is all about understanding the fundamentals of the Observable contract first.

We need to keep in mind that any given stream can only error out once, and that is exclusive with stream completion; only one of the two things can happen.

In order to recover from an error, the only way is to somehow generate a replacement stream as an alternative to the errored out stream, like it happens in the case of the catchError or retryWhen Operators.

I hope that you have enjoyed this post, if you would like to learn a lot more about RxJs, we recommend checking the RxJs In Practice Course course, where lots of useful patterns and operators are covered in much more detail.

Also, if you have some questions or comments please let me know in the comments below and I will get back to you.

To get notified of upcoming posts on RxJs and other Angular topics, I invite you to subscribe to our newsletter:

If you are just getting started learning Angular, have a look at the Angular for Beginners Course:

try {
  for (var obj in objs) {
    doSomething(obj);
  }
} catch (e) {
  handleError(e);
} finally {
  doCleanup();
}

interface Observable<T> {
  onNext(value: T) : void
  onError(error: Error) : void
  onCompleted() : void
}

var source = Rx.Observable.catch(
  get('url1'),
  get('url2'),
  get('url3'),
  getCachedVersion()
);

var subscription = source.subscribe(
  function (data) {
    // Display the data as it comes in
  }
);

var source = get('url1').catch(getCachedVersion());

var subscription = source.subscribe(
  function (data) {
    // Display the data as it comes in
  }
);

var source = get('url1').catch(function (e) {
  if (e.status === 500) {
    return cachedVersion();
  } else {
    return get('url2');
  }
});

var subscription = source.subscribe(
  function (data) {
    // Display the data as it comes in
  }
);

var source = Rx.Observable.onErrorResumeNext(
  Rx.Observable.just(42),
  Rx.Observable.throw(new Error()),
  Rx.Observable.just(56),
  Rx.Observable.throw(new Error()),
  Rx.Observable.just(78)
);

var subscription = source.subscribe(
  function (data) {
    console.log(data);
  }
);
// => 42
// => 56
// => 78

// Try three times to get the data and then give up
var source = get('url').retry(3);

var subscription = source.subscribe(
  function (data) {
    console.log(data);
  },
  function (err) {
    console.log(err);
  }
);

// Try three times to get the data and then return cached data if still fails
var source = get('url').retry(3).catch(cachedVersion());

var subscription = source.subscribe(
  function (data) {
    // Displays the data from the URL or cached data
    console.log(data);
  }
);

var source = get('url').retryWhen(
  function (attempts) {
    return attempts
      .zip(Rx.Observable.range(1, 3), function (_, i) { return i })
      .flatMap(function (i) {
        console.log('delay retry by ' + i + ' second(s)');
        return Rx.Observable.timer(i * 1000);
      });
  }
);

var subscription = source.subscribe(
  function (data) {
    // Displays the data from the URL or cached data
    console.log(data);
  }
);
// => delay retry by 1 second(s)
// => delay retry by 2 second(s)
// => Data

var socket = new WebSocket('ws://someurl', 'xmpp');

var source = Rx.Observable.from(data)
  .finally(function () { socket.close(); });

var subscription = source.subscribe(
  function (data) {
    socket.send(data);
  }
);

function DisposableWebSocket(url, protocol) {
  var socket = new WebSocket(url, protocol);

  // Create a way to close the WebSocket upon completion
  var d = Rx.Disposable.create(function () {
    socket.close();
  });

  d.socket = socket;

  return d;
}

var source = Rx.Observable.using(
  function () { return new DisposableWebSocket('ws://someurl', 'xmpp'); },
  function (d) {
    return Rx.Observable.from(data)
      .tap(function (data) { d.socket.send(data); });
  }
);

var subscription = source.subscribe();

var source1 = Rx.Observable.of(1,2,3);
var source2 = Rx.Observable.throwError(new Error('woops'));
var source3 = Rx.Observable.of(4,5,6);

var source = Rx.Observable.mergeDelayError(source1, source2, source3);

var subscription = source.subscribe(
  function (x) {
    console.log('Next: %s', x);
  },
  function (err) {
    console.log('Error: %s', err);
  },
  function () {
    console.log('Completed');
  });

// => 1
// => 2
// => 3
// => 4
// => 5
// => 6
// => Error: Error: woops