在 RxJS 中的错误拆分 observable

Question

有了 Promise，我们可以使用 .then 的变体在发生错误时拆分链。这是一个使用fetch的示例

fetch('http://website.com').then(
  // Perform some logic
  (response) => response.json().then(({ answer }) => `Your answer: ${answer}`),
  // Skip json parsing when an error occurs
  (error) => 'An error occurred :(',
).then(console.log);

这让我可以跳过响应处理逻辑，只响应原始 fetch 语句中出现的错误。 RxJS 中的类似内容可能如下所示：

Observable.fromPromise(fetch('http://website.com'))
  // if I put .catch here, the result will be piped into flatMap and map
  .flatMap(response => response.json())
  .map(({ answer }) => `Your answer: ${answer}`)
  // if I put .catch here, errors thrown in flatMap and map will also be caught
  .subscribe(console.log);

作为处于代码状态的 cmets，我不能简单地放入一个 catch 运算符，因为它的行为与我的 Promise 链不同。

我知道我可以通过涉及实现的自定义运算符来获得它，或者将捕获可观察到的错误与此合并，但这一切似乎都是相当大的矫枉过正。有没有一种简单的方法来实现承诺链的行为？

Answer 1

实际上，如果我遇到您的情况，我不会担心从flatMap 和map 捕获错误。当源 Observable 抛出错误时，它将被传播到观察者。所以我在调用 subscribe 时只使用错误处理程序（否则会重新抛出错误）：

.subscribe(console.log, err => console.log('error:', err));

请注意，当源 Observable（在您的情况下是 Promise）中发生错误时，它会作为 error 通知传播，而不是作为标准 next 通知传播。 这意味着flatMap() 和map() 绝对不会对错误消息产生任何影响。如果您使用catch() 或materialize()，则两个运算符（flatMap 和map）都必须能够处理此类数据（并且不会引发另一个错误）。

无论如何，您始终可以使用 share() 或 publish() 并进行两个不同的订阅，每个订阅只处理一种类型的信号：

let source = Observable.fromPromise(fetch('http://website.com')).publish();

source
  .subscribe(undefined, err => console.log(err));

source
  .flatMap(...)
  .map(...)
  .subscribe(console.log, () => {});

source.connect();

现在我有一个单独的观察者，只针对错误。

请注意，我必须使用() => {} 进行空回调，因此错误将被静默忽略。另请注意，当使用多播（publish() 运算符）时，内部的 Subject 可能有一些我应该注意的特定行为，但在您的用例中可能无关紧要。