ajax - Angular 2/4 & RxJS - forEach 内的订阅 - 确保流量控制在所有 Observable 完成之前不会继续

Question

目标是遍历 ID 集合，为每个 ID 进行 HTTP 调用。对于每个 ID，我都使用一个带有 get() 方法的服务，该方法返回一个 Observable。每次调用 get() 方法时，我都会订阅返回的 Observable 并尝试将结果推送到一个数组中，该数组最终会传递给不同的方法以进行新操作。

相关服务方式:

public get(departmentId: number): Observable<IDepartmentModel> {
    return super.get<IDepartmentModel>(departmentId);
}

注意:父类(super class)正在利用 Angular Http，它已经过良好测试并确认可以正常工作。逻辑的问题不在这里...

相关组件方法: 请注意在 forEach 中多次调用 departmentService.get() 调用。

 setInitialDepartmentsAssignedGridData(): void {
    this.settingsForDropdownSelectedCompanyId = this.userForm.get('defaultCompany').get('defaultCompanyId').value;

    let departments: IDepartmentModel[] = [];

    this.userService.user.getValue() //confirmed: valid user is being pulled back from the userService (logic is fine here..)
        .userCompanies.find(comp => comp.companyId === this.settingsForDropdownSelectedCompanyId) // getting a valid match here (logic is fine here..)
        .departmentIds.forEach(deptId => this.departmentService.get(deptId).first().subscribe(dept => { // getting a valid department back here (logic is fine here...)
            departments.push(dept); // HERE LIES THE PROBLEM
    }));

    this.setDepartmentsAssignedRowData(departments);
}

 setDepartmentsAssignedRowData(departments: IDepartmentModel[]): void {
    console.log('setDeptAssignedRowData called'); // confirmed: method is getting called...
    console.log(departments); // confirmed: fully-composed collection of departments logged to the console...

    departments.forEach(dept => {
        console.log(dept);
    }); // Y U NO WORK!?

    departments.map((department) => {
        console.log(department); // Y U NO WORK?
        this.departmentAssignedRowData.push({
            departmentId: department.id,
            departmentName: department.name
        });
    });

    this.departmentAssignedGridOptions.api.setRowData(this.departmentAssignedRowData);
}

问题是，虽然记录到控制台的是一个完全组合的部门对象数组，但它并不真正“存在”；传递给 setDepartmentsAssignedRowData 的是一个空数组。

我确定发生的事情是在部门数组传递给第二个方法之前异步操作尚未完成。我在网上读到的一些内容说要使用 forkJoin，但我看不出在这种情况下它会是什么样子。我还读过 concatMap 可能会起作用，但同样，在这种情况下，我不确定如何让它起作用......

在这种情况下，我如何利用 RxJS 来确保预期的、完全组合的部门数组真正准备好被传递？

感谢您提供的任何见解。非常感谢您的帮助!

Answer 1

你是对的，你需要forkJoin

let observableArray = this.userService.user.getValue()
    .userCompanies.find(comp => comp.companyId === this.settingsForDropdownSelectedCompanyId)
    .departmentIds.map(deptId => this.departmentService.get(deptId)) // map is building out an array of observables

这将是一组您希望并行生成的 http 请求可观察对象。现在您可以将此数组传递给 forkJoin。

Observable.forkJoin(...observableArray)

forkJoin 的返回将是 observableArray 的结果数组。在 observableArray 中的所有可观察对象完成之前，forkJoin 不会发送给序列中的下一个运算符(因此当所有 http 请求都完成时)

所以代码将是

let observableArray = this.userService.user.getValue()
    .userCompanies.find(comp => comp.companyId === this.settingsForDropdownSelectedCompanyId)
    .departmentIds.map(deptId => this.departmentService.get(deptId));

Observable.forkJoin(...observableArray).subscribe(res => {
    // res = [resId0, resId1, resId2, ..., resIdX];
});

---

您提到将结果传递给另一个运算符(operator)。如果该运算符是您传递数据数组(来自 forkJoin)的另一个 http 请求，那么您可以使用 flatMap运算符(operator)。

Observable.forkJoin(...observableArray)
    .flatMap(res => {
        return this.otherApi(res);
    })
    .subscribe(res => {
        // res is the result of the otherApi call
    });

flatMap 会将您的 api 请求链接在一起。所以总的来说正在发生的是

1. 并行运行可观察量数组
2. 完成后，运行第二个 api (otherApi)