RxJS: flatMap an array of values with the results of promises

user_2225054 0
Viewed 966

I have a type temperature, for example:

type Temperature = {
  source: "Sensor A",
  value: number,
}

And an observable emitting arrays of this type

temperatures$:$rxjs$Observable<Array<Temperature>>

Now I want to Perform a side effect (api call) for each of the temperature objects and append the result of those calls to the objects (when all are resolved), so a new observable would emit:

type TemperatureWithResult = {
  ...Temperature,
  result: {
    success: boolean,
    message: string
  }
}

temperaturesWithResult$:$rxjs$Observable<Array<TemperatureWithResult>>

So what I want in temperaturesWithResult is not to emit the temperature objects with the promises (resulting from the api calls) appended, but the actual results resulting from resolving those promises, so I have given a stab at this problem in the following way:

export default
  temperatures$: rxjs$Observable<Array<Temperature>>,
): rxjs$Observable<Array<TemperatureWithResult>> => {
  return temperatures$.flatMap(
    temperatures => {
      return new Promise(resolve => {
        const promisedResults = temperatures.map(temperatureObject => {

          const promisedResult = apiCall({
            temperature: temperatureObject.temperature,
          });

          return {
            ...temperatureObject,
            result: promisedResult, // this is a promise!
          };
        });

        const promises = spreadPromised.map(s => s.result);

        Promise.all(promises).then(results =>
          resolve(
            promisedResults.map((promisedResult, i) => ({
              ...promisedResult,
              result: {
                success: true, // to simplify
                message: results[i], // I update it with the actual result
              }
            })),
          ),
        );
      });
    },
  );

But this is not very readable, and there must be a simpler way.

rxjs javascript
1 Answers

The reason it becomes a bit complicated is that you have an observable emitting an array of temperatures. I don't know your specific usecase, but you should consider emitting a series of temperatures instead, i.e. using an Observable<Temperature>. It will simplify the code.

Anyway, assuming the use case you defined, this code will do it (I'm using TypeScript here but you can easily turn it into standard JS):

import { Observable } from 'rxjs/Observable';
import 'rxjs/add/observable/of';
import 'rxjs/add/observable/forkJoin';
import 'rxjs/add/operator/switchMap';
import 'rxjs/add/operator/map';

interface Temperature {
  source: string,
  value: number,
}

interface Result {
  success: boolean,
  message: string
}

interface TemperatureWithResult extends Temperature {
  result: Result
}

function apiCall(temperature: Temperature): Observable<Result> {
  // perform API call
}

function addResultsToTemperatures(temperatures: Observable<Temperature[]>): Observable<TemperatureWithResult[]> {
  // for each array of temperatures emitted, replace them with...
  return temperatures.switchMap(
    // ...the result of doing all the API calls in parallel...
    temperatures => Observable.forkJoin(
      temperatures.map(temperature => 
        apiCall(temperature)
          // ...and merging temperature and result to yield a TemperatureWithResult...
          .map(result => ({ ...temperature, result })
      )
    ))
}
user_12478530
Related