为 Observable<sourceitem[]> 中的每个项目调用 api 并将其“合并”回 Observable<targetitem[]>

Question

我有一个 Observable，其中包含用户选择的项目。每个选定的项目都用作 api 调用的源。 api 调用的结果应该“合并”到 Observable。

我似乎没有得到的东西是我如何将结果合并回来。我尝试使用扫描运算符，但“累加器”只会“增长”。

我有这个：

import {
  Observable,
  Subject,
  of ,
  from
} from 'rxjs';
import {
  map,
  merge,
  flatMap
} from 'rxjs/operators';

function apiUri(albumId: String) {
  return `https://jsonplaceholder.typicode.com/albums/${albumId}`
}


const albums$ = new Subject < String[] > ();

const responseStream = albums$.pipe(
  flatMap(albumIds => albumIds),
  flatMap(id => from(fetch(apiUri(id)))),
  flatMap(resp => resp.json()),
  /* I'm in the dark on how to "merge" it back into an Observable<String[]> */
);

responseStream.subscribe(resp => console.log(resp))
const randomCodes1 = ['5', '1', '3'];
albums$.next(randomCodes1);

带有运行示例的 Stackblitz 链接：https://stackblitz.com/edit/rxjs-flatmap-galore?&file=index.ts

Answer 1

这段代码应该允许您创建您正在寻找的responseStream。

const responseStream = albums$.pipe(
  map(albumIds => albumIds.map(id => from(fetch(apiUrl(id))).pipe(mergeMap(resp => resp.json())))),
  mergeMap(obsOfIds => forkJoin(obsOfIds)),
);

首先，让我们关注forkJoin。您可以将一个 Observable 数组传递给该运算符，该运算符将返回一个 Observable，当输入数组中的所有 Observable 都已发射时，该 Observable 将发射，并且它将发射一个数组，其中包含输入数组中每个 Observable 发射的值。

这可能是您所关注的。如何创建您需要的 Observable 数组？这就是第一个 map 运算符执行的操作。

查看该运算符时请考虑一件事。

map(albumIds => albumIds.map(id => from(fetch(apiUrl(id))).pipe(mergeMap(resp => resp.json()))))

您使用了 2 次 map，但这些不一样 map。第一个map，也就是最外层的，是Observable的map操作符。第二个map，也就是最内部的，是Array的map方法。

我已经更新了你的 stackblitz，它似乎工作正常。