我有一组需要按顺序启动的可观察对象。一旦发生错误,我需要捕获错误,记录它并继续观察。
此刻,一旦发生错误,观察者就会停止。观察者必须继续并且在出错时不重启或完成。
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
observable并单独向它们添加catch处理程序,然后一旦发生错误,它就会被正确处理并且观察者可以继续而不会完成整个连接的observable。
我们觉得应该有一个更优雅的解决方案。如果需要,我会发布我们目前的解决方案。
您可以将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; }
<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>
为了将来参考,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