在 RxJS/Angular 6 中合并相互依赖的不同 HTTP 调用

Question

我一直试图找到这个答案一段时间，但作为 RxJS 和 Angular (2+) 的新手，我正在寻找一种方法来组合两个 HTTP GET 调用的结果。一旦一个呼叫完成，就必须根据该结果进行一个新呼叫。第一次调用的结果将与第二次调用合并。最好在以对象为类型的属性中。

我一直在玩mergeMap、switchMap 和concatMap。根据文档和指南，这些功能应该做我想要实现的。但是怎么做？

在我的示例中，我有一个 ID（例如 3）。这是我的代码：

this.postsService.getPostData(postId)
.switchMap(
  postData => Observable.forkJoin(this.getUserByPostData(postData))
)
.subscribe( result => console.log(result) );

由于http.get，这两个函数（getPostData 和 getUserByPostData）都会返回 Observables

在这种情况下，尤其是由于forkJoin，我希望我的结果会被合并。但是，我检索到的结果已完全替换为第二个请求的结果（即 switchMap 函数中的那个）。

我将如何合并结果？是否也可以将第二个 HTTP 请求 (this.getUserByPostData) 的结果作为第一个请求结果的属性推送？

Answer 1

如果第二个请求依赖于第一个请求的响应中的数据，并且如果您想合并两个响应，您可以在switchMap 中使用map 运算符：

this.postsService.getPostData(postId).switchMap(
  postData => this.getUserByPostData(postData).map(
    userByPostData => ({ postData, userByPostData })
  )
).subscribe(({ postData, userByPostData })=> console.log(postData, userByPostData));

另外，如果 getPostData 返回的 observable 是基于 HTTP 的，它只会发出一次，因此永远不需要切换。你可以改用mergeMap。

Answer 2

不特定于 Angular，但我从 RxJs 的角度解决了它。

我的用例是从服务中获取令牌并在标头中使用该令牌来调用另一个服务。我想要的解决方案适合多个链式调用。

我们将有两种方法略有不同。

案例 1：

如果你有触发器/启动服务

const { forkJoin, of } = rxjs;
const { concatMap, delay, mergeMap, tap, map } = rxjs.operators;
function getAccessToken() {
  return of(`AccessToken`);
}
function getHTTP2Request(url2, reqOption) {
  console.log('reqOption received : ' + JSON.stringify(reqOption));
  return of(`Response from ${url2}`).pipe(delay(2000));
}
getAccessToken()
  .pipe(
    map((d) => {
      console.log('response in pipe', d);
      return d;
    })
  )
  .pipe(
    tap((val) => console.log(`BEFORE MAP: ${val}`)),
    map((accessToken) => {
      const requestOptions = {
        headers: {
          Authorization: `Bearer ${accessToken}`,
        },
      };
      return requestOptions;
    }),
    tap((val) => console.log(`AFTER MAP: ${val}`))
  )
  .pipe(
    concatMap((requestOptions) => {
      console.log(
        'Second : requestOptions is ' + JSON.stringify(requestOptions)
      );
      return getHTTP2Request('url 2 ', requestOptions).pipe(
        map((d) => {
          console.log('response in pipe', d);
          return d;
        })
      );
    })
  )
  .pipe(tap((val) => console.log(`AFTER All: ${JSON.stringify(val)}`)))
  .subscribe((response) => console.log('Final : ' + response));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/7.4.0/rxjs.umd.min.js"></script>

案例 2：

要么您没有触发点，要么您希望流程在多个输入上进行

const { forkJoin, of } = rxjs;
const { concatMap, delay, mergeMap, tap, map } = rxjs.operators;
function getAccessToken() {
  return of(`AccessToken`);
}
function getHTTP2Request(url2, reqOption) {
  console.log('reqOption received : ' + JSON.stringify(reqOption));
  return of(`Response from ${url2}`).pipe(delay(2000));
}
of('url1','url2')
  .pipe(
    concatMap((url) => {
      console.log('first : Sending request to ' + url);
      return getAccessToken().pipe(
        map((d) => {
          console.log('response in pipe', d);
          return d;
        })
      );
    }),
    tap((val) => console.log(`BEFORE MAP: ${val}`)),
    map((accessToken) => {
      const requestOptions = {
        headers: {
          Authorization: `Bearer ${accessToken}`,
        },
      };
      return requestOptions;
    }),
    tap((val) => console.log(`AFTER MAP:`,val))
  )
  .pipe(
    concatMap((requestOptions) => {
      console.log(
        'Second : requestOptions is ' + JSON.stringify(requestOptions)
      );
      return getHTTP2Request('url 2 ', requestOptions).pipe(
        map((d) => {
          console.log('response in pipe', d);
          return d;
        })
      );
    })
  )
  .pipe(tap((val) => console.log(`AFTER All: ${val}`)))
  .subscribe((response) => console.log('Final : ' + response));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/7.4.0/rxjs.umd.min.js"></script>

这个结构可以根据需要反复管道化。