如何在 RxJS 中使用 await答案

【问题标题】：How to use await in RxJS如何在 RxJS 中使用 await
【发布时间】：2020-02-01 14:26:36
【问题描述】：

我是 RxJS 的新手

这是我的功能

getData(id): Observable<any> {

    let accessToken = this.getAccessHeader();

    if (!accessToken) {
        // Here I want to wait till following observable completes, and returns value. Till then do not call next line methods i.e. code outside the if block

        this.getTemptoken().subscribe((res) => {
            accessToken = res.result.accessToken,
        });
    }
    const httpOptions = {
        headers: new HttpHeaders({
            'accessToken': accessToken
        })
    };
    return this.http.post(
        url,
        {
            'id': id
        }, httpOptions
    ).pipe(map((res) => {
        return res;
    }));
}

getTemptoken也是一个observable，类似的函数，调用API，返回token。我想等到getTemptoken 返回值，然后只执行下一行，即下面的代码 if block

【问题讨论】：

标签： angular async-await rxjs observable

【解决方案1】：

要在后续 observable 中使用第一个 observable 的结果，您可以在管道中使用 switchMap，例如：

of("Hello").pipe(
            switchMap(result => of(result + ", World")),
            switchMap(result => of(result + "!")),
          )
          .subscribe(console.log);
// output: Hello, World!

learnrxjs switchmap

如果您不需要上一个 observable 的结果，而只想按特定顺序触发 observable，您可以使用 concat 等待每个 observable 完成后再触发下一个 observable：

concat(
  of(1).pipe(delay(2000)),
  of(2).pipe(delay(500)),
  of(3).pipe(delay(1000))
)
.subscribe(console.log);

// output: 1 2 3

learnrxjs concat

您还可以使用toPromise()，但它是一个反模式。来看看

RxJS Observable interop with Promises and Async-Await

例如：

const source$ = Observable.interval(1000).take(3); // 0, 1, 2
// waits 3 seconds, then logs "2".
// because the observable takes 3 seconds to complete, and 
// the interval emits incremented numbers starting at 0
async function test() {
  console.log(await source$.toPromise());
}

learnrxjs topromise

【讨论】：

感谢您的快速回复，但我的要求是在一个可观察方法中，如果不存在特定数据，则调用其他可观察对象，直到停止执行第一个可观察对象

【解决方案2】：

这就是我将如何重写您的方法以按照您的意愿工作。

getData(id): Observable<any> {
  let accessToken$ = of(this.getAccessHeader())
    .pipe(
      switchMap(accessToken => accessToken ? of(accessToken) : this.getTemptoken()
        .pipe(
          map(res => res.result.accessToken)
        )
      )
    )

  return accessToken$
    .pipe(
      map(accessToken => {
        headers: new HttpHeaders({
          'accessToken': accessToken
        })
      }),
      switchMap(httpOptions => this.http.post(url, { id }, httpOptions)),
    )
}

accessToken$ 负责获取 accessToken 以防 this.getAccessHeader() 没有给我们想要的东西。这发生在 switchMap 内部，如果我们有 accessToken 我们返回 of(accessToken)，否则我们返回 this.getTemptoken()。

然后我们获取 accessToken$，将其转换为 httpOptions，并通过 switchMap 返回我们想要的 this.http.post。最后，订阅此方法的人将获得所需的 http 响应。

【讨论】：