RxJS 等待第二个 observable 然后在出错时重试原始 observable - TypeScript/Angular 2

Question

我对 Angular 2、TypeScript 和 RxJS 相当陌生，我正在创建一个利用 Salesforce Ajax 工具包连接库的简单应用程序。

我正在尝试编写一个处理程序，以便在每次调用连接库中的方法时令牌过期时进行捕获。我创建了一项服务，该服务基本上包装了连接库以使用可观察对象。例如，如果我们查看插入函数，我已经创建了自己的包装函数：

 public insert(object: sforce.SObject): Observable<any> {
   return new Observable(observer => {
   // successfully inserted the record
   let insertSuccess = (result) => {
     observer.next(result);
     observer.complete();
    }

    // An error occured inserting the record
    let insertError = (result) => {
      // This does not work yet
      if (result.faultcode.indexOf('INVALID_SESSION_ID') != -1) {
        this.refreshToken();
      }
      else {
          observer.error(result);
      }
    }

    let callback = { onSuccess: insertSuccess, onFailure: insertError };
    sforce.connection.create([object], callback);
  });
}

我有另一个刷新访问令牌的函数：

 public refreshToken(): void {
    this.loginService.login().subscribe(
        response => {

            Globals.SESSION_TOKEN = response.access_token;

            //initialize the salesforce connection 
            this.init(Globals.SESSION_TOKEN, this.loginService.AuthParams.SOAP_URL);
        },
        error => {

        }
    );
}

我基本上希望原始 insert 函数等待 refreshToken 完成。如果成功，我想再次重试相同的插入，否则我希望原始插入可观察到调用 observer.error。

我研究了 retry 和 retryWhen，但是我一直无法弄清楚如何实现它以等待 refreshToken() 函数完成。对此事的任何指导或建议将不胜感激。提前谢谢你。

Answer 1

catch 运算符接受处理错误的函数和源 Observable。这意味着如果您遇到错误，您可以确定是否要重新订阅 catch 块中的原始来源：

 public insert(object: sforce.SObject): Observable<any> {
   return new Observable(observer => {
   // successfully inserted the record
   let insertSuccess = (result) => {
     observer.next(result);
     observer.complete();
    }

    // An error occured inserting the record
    let insertError = (result) => observer.error(result);


    let callback = { onSuccess: insertSuccess, onFailure: insertError };
    sforce.connection.create([object], callback);
  }).catch((err, source) => {
     if (err.faultcode.indexOf('INVALID_SESSION_ID') != -1) {
        //This waits for the refresh to complete and then resubscribes
        //to the source
        //If the refresh errors then it will skip the resubscribe
        return this.refreshToken().flatMapTo(source);
     }
     //Non-authentication error
     return Observable.throw(err);
  });
}

然后将您的 refreshToken 函数变成如下所示：

 public refreshToken(): Observable<any> {
    return this.loginService.login()
      .tap(response => {
        Globals.SESSION_TOKEN = response.access_token;

        //initialize the salesforce connection 
        this.init(Globals.SESSION_TOKEN, this.loginService.AuthParams.SOAP_URL);
      });
}

RxJS 等待第二个 observable 然后在出错时重试原始 observable - TypeScript/Angular 2

RxJS wait for second observable then retry original observable on error - TypeScript/Angular 2

asynchronous

observable

rxjs

typescript

angular