RXJS 出错后继续 concat subscribe答案

【问题标题】：RXJS Continue with concat subscribe after errorRXJS 出错后继续 concat subscribe
【发布时间】：2018-11-30 03:31:23
【问题描述】：

我有一系列需要按顺序触发的可观察对象。一旦发生错误，我需要捕获错误、记录错误并继续观察。

此时，一旦发生错误，观察者就会停止。观察者必须继续并且不会因错误而重新启动或完成。

import * as Rx from "rxjs";
const source = [
  Rx.Observable.from("1").delay(200),
  Rx.Observable.from("2").delay(150),
  Rx.Observable.throw("error"),
  Rx.Observable.from("3").delay(124),
  Rx.Observable.from("4").delay(201),
];
let sSource = Rx.Observable.concat(...source);
sSource.subscribe((v) => {console.log(v)}, (e) => {console.log(e)});

当前输出：

1
2
error

预期输出：

1
2
error
3
4

我们能想出的唯一解决方案是预先循环遍历 source 可观察对象并分别向它们添加 catch 处理程序，然后一旦发生错误，它就会得到正确处理，并且观察者可以继续而不用完成整个连接可观察到的。

我们觉得应该有一个更优雅的解决方案。如果需要，我会发布我们目前的解决方案。

【问题讨论】：

标签： typescript rxjs rxjs5 rxjs6

【解决方案1】：

您可以将catch 运算符应用于每个源可观察对象，并可以在其中执行错误记录。像这样：

const sources = [
  Rx.Observable.from("1").delay(200),
  Rx.Observable.from("2").delay(150),
  Rx.Observable.throw("error"),
  Rx.Observable.from("3").delay(124),
  Rx.Observable.from("4").delay(201),
];
const sourcesWithCatch = sources.map(s => s.catch(e => {
  console.log(e);
  return Rx.Observable.empty();
}));
const concatted = Rx.Observable.concat(...sourcesWithCatch);
concatted.subscribe(v => console.log(v));

.as-console-wrapper { max-height: 100% !important; top: 0; }

&lt;script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"&gt;&lt;/script&gt;

【讨论】：

对不起，我离开了。这或多或少是我们最终所做的。

【解决方案2】：

为了将来参考，Rxjs 5 具有 onErrorResumeNext 函数，其行为类似于 Visual Basic On Error Resume Next 语句。

这是来自documentation的示例

var source = Rx.Observable.onErrorResumeNext(
  Rx.Observable.just(42),
  Rx.Observable.throw(new Error()),
  Rx.Observable.just(56),
  Rx.Observable.throw(new Error()),
  Rx.Observable.just(78)
);

var subscription = source.subscribe(
  data => console.log(data)
);
// => 42
// => 56
// => 78

【讨论】：