使用RxJS pipe（）将数组转换为角度数组中的异步值流

Question

type Movie = {id: string};
type FullMovie = {id: string, picture: string};

我有一个url，它返回类型为Movie的数组：

http.get(url).subscribe(res: Movie[])

我使用http.get（movie.id）为数组中的每个电影返回一个FullMovie：

http.get(movie.id).subscribe(res: FullMovie)

所以本质上，我想创建一个返回FullMovie对象流的方法，因为请求解析：getAll=（url）：可观察

getAll = (url): Observable<FullMovie> => {
  return http.get(url)
    //must pipe the array into a stream of FullMovies but not a stream of FullMovie Observables. I don't want to subscribe to each of the returned FullMovies
    //something like
   .pipe(//map(array => array.forEach(movie => return http.get(movie.id))))
}

目前我有以下解决方案，但我想一个更简洁的解决方案：

 private getFull = (queryGroup: string): Observable<TMDBMovie> =>
    new Observable<TMDBMovie>((observer) => {
      //get movie array
      this.httpGet(queryGroup).subscribe((movies) => {
        var j = 0;

        if (movies.length === 0) return observer.complete();

        //loop through elements
        movies.forEach(movie => {
          this.getById(movie.id).subscribe(
            (res) => complete(observer.next(res)),
            (error) => complete()
          );
        });
          
        }

        const complete = (arg: any = 0) => {
          if (++j === len) observer.complete();
        };
      });
    });




Answer 1

你可能想试试这些东西

getAll = (url): Observable<FullMovie> => {
  return http.get(url)
   .pipe(
      // turn the array Movie[] into a stream of Movie, i.e. an Obsevable<Movie>
      concatMap(arrayOfMovies => from(arrayOfMovies)),
      // then use mergeMap to "flatten" the various Obaservable<FullMovie> that you get calling http.get(movie.id)
      // in other words, with mergeMap, you turn a stream of Observables into a stream of the results returned when each Observable is resolved
      mergeMap(movie => http.get(movie.id))
   )
}

考虑到使用mergeMap，如上所述，您不能保证最终流的顺序与您从第一次调用中获得的Movie数组的顺序相同。这是因为每个http.get（movie.id）需要不同的时间返回，因此不能保证顺序。

如果需要保证订单，请使用concatMap而不是mergeMap（实际上concatMap是mergeMap，并发设置为1）。

如果您希望所有的http.get（movie.id）在返回结果之前完成，那么使用forkJoin而不是像这样的mergeMap

getAll = (url): Observable<FullMovie> => {
  return http.get(url)
   .pipe(
      // turn the array Movie[] into an array of Observable<Movie>
      map(arrayOfMovies => arrayOfMovies.map(movie => http.get(movie.id))),
      // then use forkJoin to resolve all the Observables in parallel
      concatMap(arrayOfObservables => forkJoin(arrayOfObservables))
   ).subscribe(
      arrayOfFullMovies => {
        // the result notified by forkJoin is an array of FullMovie objects
      }
   )
}