在RxJS中出错时拆分observable

Question

使用promises,我们可以使用变量.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);

作为代码状态中的注释,我不能简单地将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()运算符)时,主题内部可能会有一些我应该注意的特定行为,但可能在您的用例中无关紧要.