如何按顺序使用RxJS observable？ [英] How to use RxJS observables in sequential order?

问题描述

这是交易：我有一个HTTP get请求，它返回一个JSON对象列表。使用RxJS我订阅接收该列表的数据。现在，对于该列表中的每个对象，我想执行另一个HTTP请求，然后将该请求的结果放在一个数组中。

Here's the deal: I have a HTTP get request that returns a JSON list of objects. Using RxJS I subscribe to receive the data of that list. Now, for each of the objects in that list I want to perform another HTTP request and then place the results of that request in an Array.

到目前为止，我一直在能够做到这一点，但我似乎无法弄清楚如何维护数据的初始列表的顺序。这可能与整个Observable机制是异步的事实有关。

So far I've been able to do this but I can't seem to figure out how to maintain the order of the initial list with data. This probably has to do with the fact that the whole Observable mechanism is asynchronous.

这是我的代码：

    ngOnInit(): void {
    this.shiftInformationService.getShifts("2016-11-03T06:00:00Z", "2016-11-06T06:00:00Z")
        .subscribe(shifts => {
            shifts.forEach(shift => {
                this.addDataToAreaChart(shift.startDateTime, shift.endDateTime, shift.description);
            });
        });

}

addDataToAreaChart(startDate: string, endDate: string, description: string) {
    this.machineStatisticsService
        .getCumulativeMachineStateDurations(startDate, endDate)
        .subscribe(s => {
            this.areaChartData = [];
            this.areaChartData.push(new AreaChartData(startDate, endDate, description, s));
        });
}

我想要的是维持从<$ c $调用的订单在推送 areaChartData 数组中的数据时，c> shifting.forEach 循环。

What I want is to maintain the order of calling from the shifts.forEach loop when pushing the data in the areaChartData array.

有任何想法吗？帮助将不胜感激！

Any ideas? Help would be appreciated!

更新：已解决！

最终代码：

ngOnInit(): void {
    var startDate = new Date();
    startDate.setDate(startDate.getDate() - 3);

    this.shiftInformationService.getShifts(DateUtil.formatDate(startDate), DateUtil.formatDate(new Date()))
        .subscribe(shifts => {
            Observable.from(shifts)
                .concatMap((shift) => {
                    return this.machineStatisticsService
                        .getCumulativeMachineStateDurations(shift.startDateTime, shift.endDateTime)
                        .map((result) => {
                            return {
                                "addDataToAreaChartValue": result,
                                "shift": shift
                            }
                        });
                })
                .subscribe(s => {
                    this.areaChartData = [];
                    this.areaChartData.push(
                        new AreaChartData(
                            s.shift.startDateTime,
                            s.shift.endDateTime,
                            s.shift.description + ' ' + s.shift.startDateTime.slice(5, 10),
                            s.addDataToAreaChartValue
                        )
                    );
                });
        });
}

感谢Michael！

推荐答案

使用 concatMap 按顺序处理。

Use concatMap to process in sequence.

项目每个源值到一个Observable，它在输出Observable中合并，以序列化的方式等待每一个在合并下一个之前完成。

Projects each source value to an Observable which is merged in the output Observable, in a serialized fashion waiting for each one to complete before merging the next.

使用 地图 在observable中追加/转换值。

Use map to append/transform value in observable.

将给定的项目函数应用于每个发出的值通过源Observable，并将结果值作为Observable发出。

Applies a given project function to each value emitted by the source Observable, and emits the resulting values as an Observable.

所以，你需要这样做

ngOnInit(): void {
    this.shiftInformationService.getShifts("2016-11-03T06:00:00Z", "2016-11-06T06:00:00Z")
        .subscribe(shifts => {
            Rx.Observable.from(shifts) // create observable of each value in array
                .concatMap((shift) => { // process in sequence
                    return this.addDataToAreaChart(
                        shift.startDateTime, 
                        shift.endDateTime, 
                        shift.description
                    ).map((result) => {
                        return {
                           "addDataToAreaChartValue" : result, // addDataToAreaChart result
                           "shift": shift // append shift object here, so we can access it on subscribe
                        }
                    });
                })
                .subscribe(s => {
                    //this.areaChartData = []; // why??
                    this.areaChartData.push(
                        new AreaChartData(
                            s.shift.startDate, 
                            s.shift.endDate, 
                            s.shift.description, 
                            s.addDataToAreaChartValue
                        )
                    );
                });
        });
}

addDataToAreaChart(startDate: string, endDate: string, description: string) {
    return this.machineStatisticsService
        getCumulativeMachineStateDurations(startDate, endDate);
}

这篇关于如何按顺序使用RxJS observable？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！