rxjs5 合并和错误处理

问题描述

我想组合/合并多个Observable，当它们中的每一个完成时，执行一个finally 函数.merge 操作符似乎并行执行每个订阅，这正是我所需要的，但如果其中任何一个抛出错误，执行就会停止.

I would like to combine/merge multiple Observables and when each of them is completed execute a finally function. The merge operator seems to execute each subscription in parallel, which is what I need, but if any of them throws an error the execution is halted.

RxJS 版本 4 有一个运算符 mergeDelayError 应该让所有订阅一直执行直到所有订阅都完成，但是这个操作符在版本 5 中没有实现.

RxJS version 4 has an operator mergeDelayError that should keep the all subscriptions executing till all of them are completed, but this operator isn't implemented in version 5.

我应该改用不同的运营商吗?

Should I revert to a different operator?

var source1 = Rx.Observable.of(1,2,3).delay(3000);
var source2 = Rx.Observable.throw(new Error('woops'));
var source3 = Rx.Observable.of(4,5,6).delay(1000);

// Combine the 3 sources into 1

var source = Rx.Observable
  .merge(source1, source2, source3)
  .finally(() => {

    // finally is executed before all
    // subscriptions are completed.

    console.log('finally');

  });

var subscription = source.subscribe(
  x => console.log('next:', x),
  e => console.log('error:', e),
  () => console.log('completed'));

JSBin

推荐答案

我认为您可以使用 catch() 模拟相同的行为.你只需要将它附加到每个源 Observable:

I think you can simulate the same behavior by using catch(). You'll just need to append it to every source Observable:

const sources = [source1, source2, source3].map(obs =>
  obs.catch(() => Observable.empty())
);

Rx.Observable
  .merge(sources)
  .finally(...)
  ...

这篇关于rxjs5 合并和错误处理的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！