如何使用 Angular 处理 RxJS 中的错误

Question

我正在开发一个 Angular 应用程序，它显示从 RESTful API 获取的项目列表。列表的内容取决于查询。 可以通过填写输入字段、使用提交按钮或将其作为查询参数添加到 URL 来传递查询。

为确保一切正常运行并防止出现异步问题，我使用了RxJS。

现在我想知道如何处理错误，因为它们可能发生在流的中间，例如。当 HTTP 请求失败时。

查询作为输入流

这是两个Observables，它们都发送一系列查询。

// 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;
  });

获取结果

下面的Observable 会在查询已更改并从 API 获取结果时触发。

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

处理错误

现在我不知道如何处理错误，例如。 getResultsForQuery 引发错误。 我想显示错误而不是结果，但要阻止 Observable 处理新事件。

目前我已经通过从 Observable 创建两个新流来解决这个问题，一个在成功时处理结果，一个在发生错误时处理。查询无效时的响应数据包含error 属性。

成功流

// 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;
  });

错误流

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

可能的解决方案

1. 我读过throwing and catching errors，但这里的问题是流似乎在第一个错误后停止并且不会触发新事件。

2. 使用onErrorResumeNext 是described in the docs 来忽略错误并确保流在出现错误后继续。但我找不到正确“处理”错误并将其显示给最终用户的方法。

问题

在这种情况下如何处理错误是否有推荐的方法？是否需要为此创建两个流，还是建议抛出异常？

重要的是，用户知道出了什么问题，并且流在第一个错误后不会停止。

Answer 1

捕获逻辑的问题在于它有效地终止了它之前的序列，这就是为什么如果您在顶级流中使用它，它将在单个异常后停止。我建议您将 catch 逻辑包装在 flatMapLatest 中。然后，您可以捕捉内部流并转换数据以符合您下游 observers 的期望。

类似这样的：

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;
});