Как правильно работать с массивами в RxJs?

Question

[D]

packages.forEach((pack) => {
    this.packageService.getPackage(pack.id)
      .pipe(takeUntil(this.ngUnsubscribe))
      .subscribe((newPack: Package) => {
        this.verbosePackages.push(newPack);
        this.sharedPackagesService.updatePackages(this.verbosePackages);
      })
  }

getParameters(): void {
    this.verbosePackages.forEach((pack) => {
      this.packageParameters = this.packageParameters.concat(pack.packageParameters);
    })
    this.getParameterVisibilities();
  }

У меня есть две такие функции.
Как мне сделать так, чтобы функция getParameters() запускалась только после того, как в первой функции в цикле каждый элемент массива сделал запрос к серверу, получил от него ответ и добавил этот ответ в массив verbosePackages?
Можно, конечно, написать костыль, который перед запуском функции getParameters() будет сверять длину массивов packages и verbosePackages, но это явно очень плохое решение.
В целом у меня много подобных ситуаций, когда мне нужно работать с каким-то массивом, но я не знаю, пришли ли в него все необходимые данные с бекенда. Какие существуют решения, чтобы вызывать какой-то триггер, который будет уведомлять о том, что все запросы к базе данных были выполнены?

Answer 1

[Антон Швец]

getParameters(ids: (string|number)[]): Observable<MyResult[]> {
  return forkJoin(ids => ids.map(id => this.packageService.getPackage(id)))
}

Типичный параллельный запрос с получением пачки результатов.
Подписка со складыванием результатов в массив не нужна, промисы вообще не нужно никуда тащить, сразу шли в игнор странных чуваков с промисами и прорвой бессмысленного кода :)
Внутри this.packageService.getPackage возможно придется сделать обработку ошибки, т.к. если в forkJoin попадет ошибка, то сломается вся пачка, ну так же как и в случае Promise.all.

Comments

[D]

Спасибо!
Нечто подобное я и искал, ибо смешивание RxJS с промисами вызывало у меня чувство дискомфорта. Был уверен, что в RxJS должна существовать какая-то реализация.
Тем не менее, на первый взгяд, forkJoin кажется мне чем-то непонятным и сложным. Не ясно, как с помощью этого можно уйти от промисов.
Иду читать доки :)
Ещё раз спасибо.

[D]

Я понимаю, что здесь не место для кодревью, и мне нужно просто читать доки, но я был бы признателен, если бы Вы сказали, насколько грубые ошибки я допустил в примере ниже?
Так-то всё работает и я получил то, что мне нужно, сократив количество кода в несколько раз, но меня никак не покидает ощущение, что я пользуюсь инструментом не по его прямому назначение, слишком усложняю или не использую какие-то возможности в силу отсутствия опыта.

getVerbosePackages(packages: Package[]): void{
    forkJoin(this.packages.map(pack => this.packageService.getPackage(pack.id)))
      .subscribe((result) => {
        this.getPackagesData(result);
        this.getExtendedParameters();
      })
  }

 getPackagesData(packages: Package[]): void {
    forkJoin(
      packages.map(pack => this.experimentService.getLastExperimentVerbose(pack.questionnaire.id))
    ).subscribe((result) => {
      console.log(result);
    })
  }

  getExtendedParameters(): void {
    forkJoin(
      this.packageParameters.map(parameter => this.colleagueService.getParameterVisibility(parameter.id))
    ).subscribe((result) => {
      console.log(result);
    })
  }

[Антон Швец]

D, ну смотрите, в парадигме rxjs мыслят потоками.
Изначально у нас есть допустим список айдишников, создаем поток с пакеджами, как я описал выше.

getVerbosePackages(ids: (string|number)[]): Observable<Package[]> {
  return forkJoin(ids => ids.map(id => this.packageService.getPackage(id)))
}

Никаких подписок в методе быть не должно, метод возвращает обсервабл.

В компоненте будет
packages = this.getVerbosePackages(this.ids);
Потом вам надо из этих пакетов собрать данные какие-то.

getPackagesData(packages: Package[]): void {
    return forkJoin(
      packages.map(pack => this.experimentService.getLastExperimentVerbose(pack.questionnaire.id))
    )
  }

В компоненте будет

packagesData = this.packages.pipe(
  switchMap(packages => this.getPackagesData(packages))
)

В шаблоне потом где мне надо я пишу просто что-то вроде

<div *ngFor="let packageData of packagesData  | async">
</div>

и получаю внутри дива свой packageData

Как только ангуляр это попытается нарисовать, будут выполнены все запросы по цепочке. Подписки не нужны нигде.

Я в целом грубовато набросал, методы с получением данных обычно прячут в сервисы, чтобы их переиспользовать и вообще не захламлять компонент.
Сложные вещи можно выносить в свои операторы. Можно в методе принимать сразу обсервабл и отдавать другой обсервабл. И т.д. вариантов хватает.

Но общая суть - поток с исходными данными превращается в потоки с потребительскими данными и выводится на шаблон.
А в процессе этих преобразований можно ловко использовать rxjs операторы, сотни их :) Хотя в основном используется два десятка наиболее частых.

Answer 2

[GlazOtca]

Promise.all() вам в помощь

Comments

[D]

Да, промисы действительно помогли, но я никак не могу понять вот какой момент.

getVerbosePackages(): Promise<any> {
    return new Promise((resolve, reject) => {
      this.packages.forEach((pack) => {
        this.packageService.getPackage(pack.id)
        .pipe(takeUntil(this.ngUnsubscribe))
        .subscribe((pack: Package) => {
          console.log(pack);
          this.verbosePackages.push(pack);
        })
      })

      console.log('END');
      resolve(this.verbosePackages);
    })
  }

В консоли сначала выведится END, затем, по идее, должен произойти resolve промиса, и только затем начнут приходить ответы от бекенда и, соответственно, заполняться нужный мне массив.
Но на момент резолва этот массив был пустым, так как ответы от сервера ещё не пришли, но при этом промис возвращает заполненный массив.
Почему так? Почему даже после того, как произойдет резолв промиса, console.log() внутри этого промиса всё равно будет срабатывать по ходу прихода ответов от сервера.
Вероятно, я просто не правильно понимаю, что происходит в программе после того, как выполняется resolve().

[GlazOtca]

D, Вы не совсем поняли меня. Вам нужен не просто промисе, а Promise.all, который
ждёт выполнения всех промисов (или первого метода reject()). Он принимает аргументом массив промисов.

[D]

GlazOtca, ну, для Promise.all нужно написать функции-промисы. Вот я показал одну из них, но не понял, почему она работает так, как мне нужно, ведь по идее resolve() произошёл до того, как программа получила ответ от сервера.

[D]

GlazOtca, а ещё я не понимаю, зачем мне тут Promise.all?
Мне ведь для того, чтобы выполнить один промис, нужен результат предыдущего промиса. А Promise.all, если я не ошибаюсь, используется тогда, когда у нас есть несколько независимых друг от друга промисов.

[D]

(UPD)
Промисы не помогли.
В консоль-то может результат и выводится, но как я понимаю, это уже бразуер сам обновляет содержимое массива по приходе данных. Но на деле в момент выполнениня console.log() массив всё ещё пуст, так как ответ с сервера не пришёл.

[D]

GlazOtca, аааа, чёрт, кажется, я понял, что Вы имели в виду
Мне нужно каждый отдельный элемент массива обернуть в промис, а не весь массив целиком.
То есть один элемент массива делает один запрос на сервер -> резолвиться один ответ. И таким образом мы с помощью Promise.all пробегаемся по всему массиву.
Правда, я не понимаю, как это сделать, но спасибо, что сделали сдвиг в моей голове ;D

[D]

GlazOtca,
Как-то так?

getVerbosePackages(packages: Package[]): void {
    let promises = []

    packages.forEach((pack) => {
      promises.push(this.getVerbosePackage(pack));
    })

    Promise.all(promises).then((result) => {
      console.log(result);
    })
  }

  getVerbosePackage(pack: Package): Promise<any> {
    return new Promise((resolve, reject) => {
      this.packageService.getPackage(pack.id)
        .pipe(takeUntil(this.ngUnsubscribe))
        .subscribe((pack: Package) => {
          resolve(pack);
        }, error => reject(error))
    })
  }