等待多个订阅的打字稿

Question

我有一个包含 ID 列表的 Observable。对于每个 ID，我希望获得 ID 代表的对象。这样做需要我为每个对象获取一个可观察对象。在继续之前如何确保所有对象都已收到？

这是数据库中的多对多关系。

getExercises(programKey: string): Observable<Array<Exercise>> {
    let source = Observable.create(observer => {
      // ... stuff here ...
      programExercises.subscribe(programExercisesSnapshot => {
        let exercises = Array<Exercise>();
        programExercisesSnapshot.forEach(programExercise => {
          let exercise = this.getExercise(programExercise.ExerciseKey); // Returns Observable<Exercise>
          exercise.subscribe(exerciseSnapshot => exercises.push(exerciseSnapshot)); // TODO: Need to await all these subscriptions before calling observer.next()
        });
        observer.next(exercises);
      });

      return () => { }; // Dispose
    });

    return source;
}

提前致谢！

Answer 1

好吧，除了作为 Observable 的结果返回 Array 看起来有点奇怪之外，我会这样做：

getExercises(programKey: string): Observable<Array<Exercise>> {
    // ... stuff here ...
    return programExercises
        // assuming that programExercisesSnapshot is an array or can be easily converted to it
        .flatMap(programExercisesSnapshot => Observable
            .from(programExercisesSnapshot)
            .flatMap(programExercise => this.getExercise(programExercise.ExerciseKey))
            .bufferCount(programExercisesSnapshot.length));

现在让我们看看它是如何工作的。让我们从内心开始吧。

我们从数组 programExercisesSnapshot 生成 observable，然后一个一个地发出它的元素；
我们捕获这些元素，并用 this.getExercise(programExercise.ExerciseKey) 使用 flatMap();

调用返回的 observable 结果在流中 替换它们
bufferCount() 将 programExercisesSnapshot.length 个元素收集到一个数组中并将其作为结果发出。

因此，整个管道发出 this.getExercise() 次调用的结果数组。

现在，外部事物执行以下操作：

它需要 programExercises 发出的批次；

用前面描述的可观察对象发出的结果（例如数组）替换它们；

并将这些结果作为自己的结果发出。

赚了！ :)

您在原始解决方案中遗漏的另一件事是清理。当您执行 programExercises.subscribe() 时，您还需要手动取消订阅。按照我的建议去做就不需要它了——rxjs 会为你处理它。

此外，正如我一开始所说，在可观察对象中返回数组看起来有点奇怪。我希望你这样做有充分的理由。 :) 否则，您可能还需要考虑将其一一转换为可观察的发射元素。

已更新。

既然作者坦白了:)这里不需要数组是另一种方案，更简洁优雅：

getExercises(programKey: string): Observable<Exercise> { // ... stuff here ... return programExercises // assuming that programExercisesSnapshot is an array or can be easily converted to it .flatMap(programExercisesSnapshot => Observable.from(programExercisesSnapshot)) .flatMap(programExercise => this.getExercise(programExercise.ExerciseKey));

等待多个订阅的打字稿

Typescript awaiting multiple subscriptions

observable

rxjs

typescript

angularfire2

angular