我有一组模型,我想要映射并对它们进行异步操作,以润滑该模型的属性,然后最终订阅现在水合的原始模型列表.

恩.伪代码.需要一些帮助填补空白和/或重构代码,这样才有意义

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;
})

解决方法:

主要技巧是让你的承诺实际返回修改后的人并从你的地图操作返回承诺.这给了Rx它可以等待地理定位何时完成的东西.

假设person.save()是同步的,我认为没有理由不在你获得地理定位结果后立即调用它.以下是您可以将它串在一起的方法,使用merge(maxConcurrent)来限制飞行中的同时地理定位请求的数量.

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
    });