What is a way to execute a series of requests using rxjs similar to forkJoin and combineLatest, without needing to wait for all requests to finish before viewing the results?

Question

What is a way to execute a series of requests using rxjs similar to forkJoin and combineLatest, without needing to wait for all requests to finish before viewing the results?

Consider you have a list of web addresses:

urls: string[]

You create a set of requests (in this instance, utilizing Angular's HTTPClient.get which gives back an Observable)

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

The objective is to run these requests concurrently while receiving responses gradually. In simple terms, if there's something like films$: Observable<Film[]>, I intend for films$ to update step by step with each new response received.

To simulate this scenario, adjust the code snippet above as follows

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

With the updated array of Observables mentioned above, data from each request should be retrieved sequentially since they are not all requested simultaneously. Bear in mind that this is just a representation of staggered data arrival times from individual requests. The requests themselves are executed concurrently.

The aim is to refresh the items in films$ every time any of the requests emit a value.

...

javascript angular typescript rxjs

Answer 1

Answer №1

If you're looking for a solution, consider utilizing the scan operator:


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), [])
);

The sequence of actions is as follows:

from will emit URLs one by one
mergeMap will subscribe to "makeRequest" and emit the result into the stream
scan will accumulate results into an array and emit each time a new emission is received

To maintain order, using combineLatest could be beneficial as it emits an array in the same order as the input observables. Each observable can start with startWith(undefined), followed by filtering 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))
);

Answer 2

If you're looking for a solution, consider utilizing the scan operator:


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), [])
);

The sequence of actions is as follows:

from will emit URLs one by one
mergeMap will subscribe to "makeRequest" and emit the result into the stream
scan will accumulate results into an array and emit each time a new emission is received

To maintain order, using combineLatest could be beneficial as it emits an array in the same order as the input observables. Each observable can start with startWith(undefined), followed by filtering 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))
);

What is a way to execute a series of requests using rxjs similar to forkJoin and combineLatest, without needing to wait for all requests to finish before viewing the results?

Answer №1

Similar questions

Unrestricted Angular Audio Playback without CORS Restrictions

Adjusting the Connection header in a jQuery ajax request

Express throwing module errors

Press the button to reveal the hidden Side Menu as it gracefully slides out

A function that takes in a type identifier and a portion of a type, and then outputs the full type

I am looking to refund the sum

The button event listener in React fails to trigger without a page refresh

Ways to efficiently incorporate data into App.vue from the constructor

Problem detected in id modification

Does jqgrid navgrid have an event called "on Refresh"?

What could be the reason for my code generating the error [$http:badreq]?

Displaying server errors in an Angular componentIn this tutorial, we

Refresh the dropdown menu selection to the default option

Guidance on using an array to filter an object in Javascript

Utilizing props in makeStyles for React styling

The term "Highcharts does not exist"

What allows us to create an instance of a generic class even without defining the generic type parameter?

Using a static string in Javascript yields no issues, whereas working with variables can sometimes cause problems

Tips for showing solely the current page number within angular pagination

Using Javascript to dynamically add rows to a table, however they mysteriously appear and disappear soon