How to handle errors in RxJS using Angular

Question

I'm working on an Angular application which shows a list of items fetched from a RESTful API. The content of the list depends on a query. The query can be passed either by filling in an input field, using a submit button or by adding it to the URL as query parameter.

To make sure everything runs in order and prevent asynchronous issues, I’m using RxJS.

Now I’m wondering how I can handle errors, since they can occur in the middle of a stream, eg. when the HTTP-request fails.

The query as input stream

These are two Observables, which both send a sequence of queries.

// first observable for the submit button
submitFormObservable = $scope.$createObservableFunction('search');

// second observable for when the input value changes
inputObservable = $scope.$toObservable('query')
  .map(function (change) {
    return change.newValue;
  });

Fetching for the results

The Observable below triggers when the query has been changed and fetches the results from the API.

var mainStream = Rx.Observable.merge([submitFormObservable, inputObservable])
  .where(function (query) {
    return query && query.length > 0;
  })
  .debounce(400)
  .distinctUntilChanged()
  .select(getResultsForQuery)
  .switchLatest();

Handling errors

Now I don’t know how to handle errors when eg. the getResultsForQuery throws an error. I would like to display the error instead of the results, but prevent the Observable from handling new events.

Currently I've solved it by creating two new streams from the Observable, one to handle the result when successful and one when an error occurs. The response data for when the query was invalid contains an error property.

Success stream

// stream containing the album information from LastFm
mainStream
  .filter(function (response) {
    return !response.data.error;
  })
  .map(function (response) {
    return response.data.result;
  })
  .subscribe(function (result) {
    $scope.error = undefined;
    $scope.result = result;
  });

Error stream

mainStream
  .filter(function (response) {
      return response.data.error;
    })
   .map(function (response) {
      return response.data;
  });
  .subscribe(function (error) {
    $scope.result = [];
    $scope.error = error;
  });

Possible solutions

1. I've read about throwing and catching errors, but the problem here is the stream seems to stop after the first error and doesn't trigger new events.

2. Using onErrorResumeNext is described in the docs to ignore errors and make sure the stream continues after an error. But I can't find a way to correctly 'handle' the error and show it to the end user.

Question

Is there a recommended approach on how to handle errors in a situation like this? Is it necessary to create two streams for this, or is it recommended to throw an exception?

It's important that the user knows something went wrong, and the stream doesn't stop after the first error.

Answer 1

The issue with catching logic is that it effectively terminates the sequence before it, which is why if you use it in your top level stream it will stop after a single exception. I would recommend that you wrap your catch logic inside of a flatMapLatest. Then you can catch on the inner stream and transform the data to conform with what your downstream observers expect.

Something like this:

var mainStream = Rx.Observable.merge([submitFormObservable, inputObservable])
  .where(function (query) {
    return query && query.length > 0;
  })
  .debounce(400)
  .distinctUntilChanged()
  .selectSwitch(function(input) {
    return getResultsForQuery(input)
            .map(function(response) {
              return {result : response.data.result};
            })
            .catch(function(e) {
              return Rx.Observable.just({error : error});
            });
  });

mainStream.subscribe(function(r) {
  $scope.result = r.result || [];
  $scope.error = r.error;
});