Use RxJS pipe() to turn array into stream of asynchronous values in Angular

Question

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

I have a url that returns an array of type Movie:

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

I use http.get(movie.id) for each movie in the array returning a FullMovie:

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

so in essence I want to create a method that returns a stream of FullMovie objects, as the requests resolve: getAll = (url): Observable<FullMovie>

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))))
}

At the moment I have the following solution that works but I want to a more concise solution:

 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();
        };
      });
    });

EDIT:

This works

newGetFull = (queryGroup: string) =>
    this.httpGet(queryGroup)
      .pipe(concatMap((arr) => from(arr)))
      .pipe(
        mergeMap((movie) => this.getById(movie.id).pipe(catchError(() => of())))
      );

Picci · Accepted Answer · 2021-12-12 18:53:25Z

3

You may want to try something along these lines

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))
   )
}

Consider that using mergeMap as above you do not have guarantee that the final stream will have the same order as the array of Movies you get from the first call. This is because each http.get(movie.id) can take different time to return and therefore the order is not guaranteed.

If you need to guarantee the order, use concatMap rather than mergeMap (actually concatMap is mergeMap with concurrency set to 1).

If you want all the http.get(movie.id) to complete before returning the result, then use forkJoin rather than mergeMap like this

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
      }
   )
}

answered Dec 12, 2021 at 18:53

Picci

17.8k16 gold badges80 silver badges125 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

PiwiTheKiwi Over a year ago

This works perfectly thanks! Just one more thing, If one of the http requests returns a 404 error the whole Observable stops, where do I use catchError() so it will continue since the code in my edit doesn't work in the right way?

PiwiTheKiwi Over a year ago

Never mind I added it in the correct place thanks

Collectives™ on Stack Overflow

Use RxJS pipe() to turn array into stream of asynchronous values in Angular

EDIT:

1 Answer 1

2 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

EDIT:

1 Answer 1

2 Comments

Your Answer

Sign up or log in

Post as a guest

Related