How to run an array of requests with rxjs like forkJoin and combineLatest but without having to wait for ALL to complete before seeing the results?

user_2678218 0
Viewed 920

So imagine you have an array of URLs:

urls: string[]

You make a collection of requests (in this example I am using Angular's HTTPClient.get which returns an Observable)

const requests = urls.map((url, index) => this.http.get<Film>(url)

Now I want to execute this requests concurrently but not wait for all response to see everything. In other words, if I have something like films$: Observable<Film[]>, I want films$ to update gradually every time a response arrives.

Now to simulate this, you can update the requests above into something like this

const requests = urls.map((url, index) => this.http.get<Film>(url).pipe(delay((index + 1)* 1000))

With the above array of Observables you should get data from each request one by one since they aren't requested at the same time. Note that this is just faking the different times of arrival of data from the individual requests. The requests itself should be done concurrently.

The goal is to update the elements in films$ every time value emitted by any of the requests.

So before I had something like this when I misunderstood how combineLatest works

let films$: Observable<Film[]> = of([]);
const requests = urls.map(url => this.http.get<Film>(url)
.pipe(
  take(1),
 // Without this handling, the respective observable does not emit a value and you need ALL of the Observables to emit a value before combineLatest gives you results.
 // rxjs EMPTY short circuits the function as documented. handle the null elements on the template with *ngIf. 
  catchError(()=> of(null))
));
// Expect a value like [{...film1}, null, {...film2}] for when one of the URL's are invalid for example.
films$ = combineLatest(requests);

I was expecting the above code to update films$ gradually, overlooking a part of the documentation

To ensure the output array always has the same length, combineLatest will actually wait for all input Observables to emit at least once, before it starts emitting results.

Which is not what I was looking for.

If there is an rxjs operator or function that can achieve what I am looking for, I can have a cleaner template with simply utilizing the async pipe and not having to handle null values and failed requests.

I have also tried

this.films$ = from(urls).pipe(mergeMap(url => this.http.get<Film>(url)));

and

this.films$ = from(requests).pipe(mergeAll());

which isn't right because the returned value type is Observable<Film> instead of Observable<Film[]> that I can use on the template with *ngFor="let film of films$ | async". Instead, if I subscribe to it, it's as if I'm listening to a socket for one record, getting updates realtime (the individual responses coming in). I can manually subscribe to any of the two and make an Array.push to a separate property films: Film[] for example, but that defeats the purpose (use Observable on template with async pipe).

rxjs angular typescript javascript
1 Answers

The scan operator will work nicely for you here:


const makeRequest = url => this.http.get<Film>(url).pipe(
  catchError(() => EMPTY))
);

films$: Observable<Film[]> = from(urls).pipe(
  mergeMap(url => makeRequest(url)),
  scan((films, film) => films.concat(film), [])
);

Flow:

  • from emits urls one at a time
  • mergeMap subscribes to "makeRequest" and emits result into stream
  • scan accumulates results into array and emits each time a new emission is received

To preserve order, I would probably use combineLatest since it emits an array in the same order as the input observables. We can start each observable with startWith(undefined), then filter out the undefined items:


const requests = urls.map(url => this.http.get<Film>(url).pipe(startWith(undefined));

films$: Observable<Film[]> = combineLatest(requests).pipe(
  map(films => films.filter(f => !!f))
);
user_18583570
Related