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

Question

我在 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

Answer 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 )
    );
}

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

How to continue after recieving an error on a stream in RxJs

node.js

rxjs

typescript