Using RxJs to hydrate model properties async

Question

I have an array of models that I want to map over and fire async operations on them to hydrate a property of that model and then finally subscribe to the original list of models that are now hydrated.

ex. pseudo code. Need some assistance filling in the blanks and/or restructuring the code so it makes more sense

var people = [{
    name: 'Jon',
    location: 'New York, NY'
}, {
    name: 'Joe',
    location: null
}, {
    name: 'Tom',
    location: 'San Francisco, CA'
}];


var source = Rx.Observable.from(people);

source.map(function(person){
    // Take current person.location and run it through geolocating service. I also need to throttle() the requests to the geolocating service.
    //ex. 
    geocorder.geocode(person.location).then(function(result){
        person.location = result;
    });

    return person;
})
.map(function(people){
    //Now that model has been updated

    person.save();
})
.subscribe(function(people){
    //Once all the saving of people have completed. Show UI message;
})

Answer 1

The main trick is to have your promise actually return the modified person and return the promise from your map operation. This gives Rx something it can wait upon to know when the geolocating is finished.

Assuming person.save() is synchronous, I see no reason not to call it right after you get your geolocation results. Here's how you might string it together, using merge(maxConcurrent) to limit the number of simultaneous geolocation requests in flight.

source
    .map(function (person) {
        // perform geolocating...
        // return the geolocated person
        // defer the call so that it is not actually performed until
        // merge() subscribes to it.
        return Rx.Observable.defer(function () {
            return geocorder.geocode(person.location).then(function (r) {
                person.location = r;
                person.save(); // any reason not to save here?
                return person;
            });
        });
    })
    .merge(2) // no more than 2 geocode requests outstanding at any given time
    .toArray() // combine the individual people back into an array
    .subscribe(function (people) {
        // do something with the people
    });