嗨,我有一个 (NgRx) Action,它触发了一个函数(应该返回一个Observable),该函数具有一个循环,每次循环我都需要发出(next)一个值,在循环中获取数据的函数返回一个 Observable,当所有 observable 完成时,我想完成具有循环的 observable:
以下是代码:
public outerFunction(
collections: someObject
): Observable<someOtherObject> {
const outerObservable$ = new Observable<Observable<someOtherObject>>(
(observer) => {
const collKeys = Object.keys(collections);
for (const id of collKeys) {
if (collections[id]) {
const innerObs$ = this.functionThatReturnsObs(
collections[id]
)
observer.next(innerObs$);
}
// observer.complete(); <-- If I complete here the stream is interrupted
}
}
);
return outerObservable$.pipe(mergeAll());
}