combining results from multiple rxjs observables

Question

I have an autocomplete input which, as the user types, fetches data from multiple endpoints, such as:

//service call to fetch data and return as single observable
getAutocompleteSuggestions() {
    const subs$ = [
        this.http.get(endpoint1),
        this.http.get(endpoint2),
        this.http.get(endpoint3)
    ];

    return Observable.forkJoin(...subs$);
}

Each of these endpoints returns data of the form:

{ data: [], status: xyz }

I would like to use switchmap as I want to only show results from the final call, and have tried the following:

   this.getAutocompleteSuggestions(query)
          .switchMap(res => {
             return res.data;
           })
          .subscribe((results: any) => {
            this.results = results;
          });

But the 'res' in the switchmap is an array, any idea how results can contain a single array containing the data from the response of any number of observables?

That's what forkJoin does. It combines the results into a single array. Maybe I don't understand your question though... can you clarify? — jme11, Mar 07 '18 at 14:02
forkjoin produces a result which is an array, so i want to flatten that array, transform it (such as through map()) and cancel previous requests — Ben Taliadoros, Mar 07 '18 at 14:06

Robin Dijkhof · Accepted Answer · 2018-03-07T14:24:43.123

5

I don't fully understand what you want, but I think this is it:

$filter: Subject<string> = new Subject(); //I guess we have some value to filter by??

Push a value to the subject:

this.$filter.next(myNewValue);

In the constructor or init:

this.$filter
   .switchMap(filterValue => { //Get the values when filter changes
       subs$ = [
         this.http.get(endpoint1 + filterValue),
         this.http.get(endpoint2 + filterValue),
         this.http.get(endpoint3 + filterValue)
       ];

       return Observable.forkJoin(...subs$);
   })
   .map(results => { //now map you array which contains the results
      let finalResult = [];
      results.forEach(result => {
          finalResult = finalResult.concat(result.data)
      })
      return final;
   })
   .subscribe(); //Do with it what you want

The entire steam will be executed again when we put a new value into our subject. SwitchMap will cancel all ready requests if there are any.

edited Mar 07 '18 at 14:24

answered Mar 07 '18 at 14:05

Robin Dijkhof

18,665
11
65
116

Interesting, so the forkjoin is inside the switchMap? is it possible to chain it so the switchmap takes the response of my requests/observables? – Ben Taliadoros Mar 07 '18 at 14:08
1

@BenTaliadoros at that point it will be too late since `switchMap` won't be able to cancel those observables. – Explosion Pills Mar 07 '18 at 14:10
Not cancel, just emit the last result – Ben Taliadoros Mar 07 '18 at 14:11
I think maybe I'm confused around the chaining element, I'm seeing examples such as: Rx.Observable.of('some_url') .switchMap(url => this.http.get(url)), so maybe i have to make a blank observable and chain the switchmap to that can call the service in there? – Ben Taliadoros Mar 07 '18 at 14:14
1

SwitchMap can change the observable steam from a value to a new value which comes from another stream. In that small example: string to response from a new http stream. This will only work once because Observable.of will never emit multiple values. If you want another result you need to create and call the stream again. In you question the value comes from a autocomplete. This value changes so we put it in a subject. The difference between Observable.of and the subject is the subject can emit more than once. Since the subject emits multiple values we use switchmap to cancel previous request. – Robin Dijkhof Mar 07 '18 at 14:22
So I used your answer to refactor my code, see https://stackoverflow.com/questions/49191962/convert-rjxs-map-and-flatten-to-flatmap for an updated question – Ben Taliadoros Mar 09 '18 at 10:45

score 1 · Answer 2 · answered Mar 07 '18 at 14:00

I have used something similar to this, and worked well so far:

        let endpoint1 = this.http.get(endpoint1);
        let endpoint2 = this.http.get(endpoint2);
        let endpoint3 = this.http.get(endpoint3);

        forkJoin([endpoint1, endpoint2, endpoint3])
          .subscribe(
            results => {
              const endpoint1Result = results[0].map(data => data);
              const endpoint2Result = results[1].map(data => data);
              const endpoint3Result = results[2].map(data => data);

              this.results = [...endpoint1Result, ...endpoint2Result, ...endpoint3Result];
            },
            error => {
              console.error(error);
            }
          );

Obviously is a pretty simple example, you'll be able to handle better the results to suit your needs.

Yea I know I can do this, but it doesn't use switchMap, thanks though — Ben Taliadoros, Mar 07 '18 at 14:02

combining results from multiple rxjs observables

2 Answers2