javascript - RXjs - 即使发生错误也继续收听

Question

我有两个(多个)异步函数包装在一个 Observable 中，我想一起运行它们并检查是否有错误或已完成。这是我所做的:

var observables = [];

observables.push(new Observable((observer:any) => {
    async1(options, (error, info) => {
        if (error) {
            observer.error(error);
        } else {
            observer.next(info);
            observer.complete();
        }
    });
}))

observables.push(new Observable((observer:any) => {
    async2(options, (error, info) => {
        if (error) {
            observer.error(error);
        } else {
            observer.next(info);
            observer.complete();
        }
    });
}))

Observable.forkJoin(observables).subscribe(
    data => {
        console.log(data);
    },
    error => {
        console.log(error);
    }
)

我的问题来了...当两个异步函数都成功完成时，它会调用 data =>{} 并在一个数组中返回两个结果。

如果这两个函数之一失败，它将调用一次 error =>{}，仅此而已。我想听每一个错误，我该怎么做？

Answer 1

合并多个流的运算符的默认行为是在其中一个流发出错误通知后立即退出。之所以如此，是因为根据 Rx 语法，错误是最终的，因此通常假设运算符返回的流必须立即失败。

这里一个简单的解决方案是取消 error 通知，并将其替换为插入到 next 通知中的错误数据结构。

所以像这样:

observables.push(new Observable((observer:any) => {
    async1(options, (error, info) => {
        if (error) {
            observer.next({error});
        } else {
            observer.next({info});
            observer.complete();
        }
    });
}))

然后在您的订阅中:

Observable.forkJoin(observables).subscribe(
    arrayData => arrayData.forEach(data => data.info? {
        console.log(data.info);
    } : {
        console.log(data.error);
    })
)