出错后不要关闭 Observable

Question

我知道这个问题看起来很简单，但是网上关于 Observables 和错误处理的资源并不是很好（或者我可能只是不擅长搜索）。

我有一个 http 请求，它返回一个Observable的Response，这些 Responses 可以包含数据或错误消息。如果它包含数据我想提取它并解析它，如果它包含错误消息我想跳过所有其他操作符（关于解析）并在订阅者中执行错误函数。

我可以做所有这些事情并抛出错误：

http.get(...).
...
.do(res=>{
  if(res.error) throw new Error(res.error.message);
  return res;
})

它可以工作，它跳过所有运算符并执行错误函数。问题是，在错误发生后，订阅者停止并且不再接受数据。

如果我在错误发生后分析订阅者，我会注意到属性closed和isStopped都设置为true。我想防止这种情况，我想在出现错误后也保持 Observable 处于活动状态。我该怎么做？

谢谢

Answer 1

正如前面的答案中提到的，这是标准行为，由Rxjs可观察的合同保证。如果你想逃避合同，你可以materialize源，在其中，而不是处理的消息通过你的观察到流的情况下，你会处理元的消息（称为通知，您所期望的三类，next，error，complete） . 元消息没有契约，这让你有责任创建错误，手动完成消息，或用于dematerialize返回到正常行为。

一般来说，observables 非常适合数据流，而控制流通常是艰巨的（跳跃、循环、条件分支等）。

请参阅Notification此处的文档：http : //reactivex.io/rxjs/class/es6/Notification.js~Notification.html。

请参阅materialize此处的文档：http : //reactivex.io/rxjs/class/es6/Observable.js~Observable.html#instance-method-materialize

因此，如果您想在不停止源 observable 的情况下处理错误，您可以执行以下操作：

function process(source, success, error) {
  return source.materialize()
    .map(function(notification){
      const value = notification.value; // that is the actual message data
      if (notification.kind === 'N') {
        // that is a next message, we keep it a next message
        return Notification.createNext(success(value));
      }
      if (notification.kind === 'E') {
        // that is a error message, we turned into a next message
        return Notification.createNext(error(value));
      }
      if (notification.kind === 'C') {
        // that is a completed message, we keep it a completed message
        return Notification.createComplete();
      }
    })
   .dematerialize()
}

然后，您可以将此函数与源 observable 一起使用，并将处理流值的成功和错误函数传递给它。如您所见，这里的技术是将错误消息转换为正常消息，然后恢复正常。

这是从我的脑后完成的，未经测试，所以如果这对你有帮助，请让我更新。我在 Rxjs v4 中多次使用了类似的技术，我相信它应该以直接的方式转换为 Rxjs v5。