如何按顺序使用 RxJS 可观察对象？

Question

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

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

这是我的代码：

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

我想要的是在推送areaChartData数组中的数据时保持shifts.forEach循环的调用顺序。

有什么想法吗？帮助将不胜感激！

更新：已解决！

最终代码：

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

感谢迈克尔！

Answer 1

使用concatMap依次处理。

将每个源值投影到一个 Observable 中，该 Observable 合并到输出 Observable 中，以序列化的方式等待每个源值完成，然后再合并下一个。

使用map 在可观察对象中附加/转换值。

将给定的项目函数应用于源 Observable 发出的每个值，并将结果值作为 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);
}