在 RxJs 中的流上收到错误后如何继续答案

【问题标题】：How to continue after recieving an error on a stream in RxJs在 RxJs 中的流上收到错误后如何继续
【发布时间】：2016-01-17 22:15:43
【问题描述】：

我在 Node.js 中有一项服务，可以检索邮件服务器的电子邮件地址：

export interface IHasEmail {
    email: string;
}

export interface IHasMailServer extends IHasEmail {
    mailServers: Array<string>
}

export interface IMailServerService {
    lookupMailServers( contact: IHasEmail ): RX.Observable<IHasMailServer>;
}

我有一组联系人 (IHasEmail)，我想检索整个列表的邮件服务器。到目前为止，我有这个：

checkContacts( contacts: Array<contracts.IContact> ): void {

    Rx.Observable.from(contacts).flatMap<contracts.IHasMailServer>( (contact) => {
        return this._mailServerService.lookupMailServers( contact )
    } ).subscribe(
        (result) => this.handleResult(result),
        (error) => this.handleError( error )
    );
}

这可以正常工作，直到从对 lookupMailServer 的调用之一返回错误。此时，handleResults 不再返回任何结果。

我想记录错误但继续从其他联系人那里收到成功的结果。我该怎么做？

这个例子的代码可以在这里看到：

https://github.com/Roaders/contact-list-checker/tree/4dd0146449a4be6e703a9b60034231cc4c4bdea3

【问题讨论】：

您需要明确处理错误。创建一个新的 observable 来处理它，并将它映射到类似{error: Error, result: IHasMailServer} 的东西。然后使用这个 observable，通过error == null 过滤它并进行日志记录。通过error != null 过滤它，然后在你的信息流中做任何事情。

标签： node.js typescript rxjs

【解决方案1】：

Observables 的终端事件除外。如果一个内部错误，它将冒泡到外部Observable 并一直转发到下游。您需要 catch 或 retry 中间异常，就像在同步代码中一样。

checkContacts( contacts: Array<contracts.IContact> ): void {

    Rx.Observable.from(contacts).flatMap<contracts.IHasMailServer>( (contact) => {
        return this._mailServerService.lookupMailServers( contact )
                 //If it is a cold source than retrying the method will
                 //re-invoke the method, just in case you suffered something like a transient network error.
                 .retry(3)
                 .catch((err) => {
                   //Log the error message here
                   //Then return an empty Observable which will get removed
                   //during the subsequent flattening.
                   return isFatal(err) ? Rx.Observable.throw(err) : Rx.Observable.empty();
                 });
    } ).subscribe(
        (result) => this.handleResult(result),
        (error) => this.handleError( error )
    );
}

【讨论】：

完美，非常感谢。我确实删除了 source 参数，因为它产生了 Typescript 编译错误。效果很好。