Машинопись, ожидающая нескольких подписей

Question

У меня есть Observable, содержащий список идентификаторов. Для каждого идентификатора я хочу получить объект, который представляет 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 выглядит немного странно, это как бы я это сделать:

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

Теперь давайте посмотрим, как это должно работать. Начнем с внутренней вещи.

1. мы генерируем наблюдаемый из массива programExercisesSnapshot, который затем испускает элементы один за другим;
2. мы улавливаем эти элементы и заменяем их в потоке с результатами наблюдений, возвращаемыми this.getExercise(programExercise.ExerciseKey) звонками с использованием flatMap();
3. bufferCount() собирает programExercisesSnapshot.length элементов в один массив и испускает их в результате.

Таким образом, весь этот трубопровод излучает массивы результатов вызовов this.getExercise().

Теперь внешняя вещь делает следующее:

1. он принимает партии, испускаемые programExercises;
2. заменяет их результатами (например, массивами), испускаемыми ранее описанными наблюдаемыми;
3. и испускает эти результаты как свои собственные.

Прибыль! :)

Еще одна вещь, которую вы пропустили в своем оригинальном решении, - это очистка. Когда вы делаете 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));