使用 forkJoin 在 foreach 中等待可观察的订阅

Question

我试图在我的组件中填充一个名为约定的数组，它是一个约定数组。

每个组织都有一个合同列表，每个合同都有一个约定 id，有了这个 id，我就得到了约定。

我使用 getOrganizationForUser 获取当前组织，然后获取合同列表。 然后我使用合约中的约定 id 调用第二个 API 来获取约定。

目前，我的代码如下所示：

public getOrganizationForUser(): Observable<Organization> {
        return this.httpClient
            .get<Organization>(`${c.serviceBaseUrl.sp}/organizationByUser`)
            .pipe(catchError((err, source) => this.responseHandler.onCatch(err, source)));
    }

 public getById(id: number) {
        return this.httpClient
            .get<Convention>(`${c.serviceBaseUrl.sp}/conventions/` + id)
            .pipe(catchError((err, source) => this.responseHandler.onCatch(err, source)));
    }

ngOnInit() {
    this.OrganizationService.getOrganizationForUser().subscribe((organization: Organization) => {
        organization.contracts.forEach((contract) => {
            this.conventionService.getById(contract.conventionId).subscribe((convention: Convention) => {
                this.conventions.push(convention);
            })
        })
   })
}

我知道我可以创建一个 observable 数组，并使用 Observable.forkJoin() 等待所有这些异步调用完成，但我希望能够定义订阅回调 每个调用的函数，因为我需要对该过程的引用。关于如何解决这个问题的任何想法？

我试过这个函数，但总是返回了解

getTasksForEachProcess(): Observable<Array<any>> {
    let tasksObservables = this.organizationService.getOrganizationForUser().pipe(map((organization: Organization) => {
        organization.contractOrganizations.map(contract => {
            return this.conventionService.getById(contract.conventionId).subscribe(convention =>
                this.conventions.push(convention)
            )
        });
    })
    );
    return forkJoin(tasksObservables);
};

ngOnInit() {
    this.getTasksForEachProcess().subscribe(item => {
        console.log(item);
    }
}

Answer 1

首先我不确定你真正想要实现什么，因为我不明白你的意思

我希望能够为每个人定义订阅回调函数 调用次数，因为我需要对进程的引用

无论如何，在你描述的那种情况下，我会做这样的事情

public getOrganizationForUser(): Observable<Organization> {
        return this.httpClient
            .get<Organization>(`${c.serviceBaseUrl.sp}/organizationByUser`)
            .pipe(catchError((err, source) => this.responseHandler.onCatch(err, source)));
    }

 public getById(id: number) {
        return this.httpClient
            .get<Convention>(`${c.serviceBaseUrl.sp}/conventions/` + id)
            .pipe(catchError((err, source) => this.responseHandler.onCatch(err, source)));
    }

ngOnInit() {
    const obsOfObservables = this.OrganizationService.getOrganizationForUser()
    .pipe(
       map(organization => organization.contracts),
       map(contracts => contracts.map(contract => this.conventionService.getById(contract.conventionId)))
    );

    obsOfObservables
    .pipe(
       switchMap(conventionObservables => forkJoin(conventionObservables))
    )
    .subscribe(
       conventions => { // do stuff with conventions }
    )
}

这里的重点如下。

通过getOrganizationForUser()，你会得到一个发出组织的 Observable。您做的第一件事就是将 Observable 发出的对象转换为具有第一个 map 运算符的合约数组。

第二个map 运算符将合约数组转换为约定的可观察数组。为了执行这个转换，我们在 Observable 的 map 运算符中使用数组的 map 方法。这可能有点令人困惑，但值得理解。

如果我们停在这里，我们所拥有的是obsOfObservables，即一个发出一个 Observable 数组的 Observable。

然后，我们将 obsOfObservables 发出的 Observable 数组传递给 forkJoin 函数，该函数本身返回一个 Observable。由于我们实际上对forkJoin 返回的 Observable 所通知的内容感兴趣，即我们对约定感兴趣，所以我们需要从第一个 Observable 切换到第二个 Observable，这样就完成了通过switchMap 运营商。

最终结果是一个 Observable，它返回一个约定数组。考虑到添加常量 obsOfObservables 是为了澄清推理，它完全没有必要（正如 Barney Panofsky 所说）。

我没有模拟整个事情，所以我希望我没有插入错误，但或多或​​少这是我在这种情况下会使用的思考过程。

最后一点，当您在subscibe 中包含subscribe 时，通常要保持怀疑。

Answer 2

我同意 Picci 思考答案的逻辑。这里与他的提议略有不同，虽然我和他一样没有严格测试过，可能会有一些错误。

这样做的逻辑是，您最终想要的是一个约定数组，而从 observables 生成一个数组就是 'zip' 所做的。所以这是流程：

• 获取一个组织，然后使用 rxjs 的 from 从 organization.contracts 数组中创建一个可观察数据流。
• 该流中的每个项目都将是一个合约，然后将转换（使用map）为基于使用contract.conventionId 属性的API 查找的约定。
• 整个生成的约定的 observable 流最终将通过包装 zip 转换回数组，并作为可订阅的 observable 交付，从而产生所需的约定数组。

代码如下：

ngOnInit() {
    zip( this.OrganizationService.getOrganizationForUser()
        .pipe(
            map((organization: Organization) => 
                from(organization.contracts).pipe(
                    map(contract => this.conventionService.getById(contract.conventionId))
                )
            )
        )
    )
    .subscribe((conventions: Convention[]) => this.conventions = conventions)
}