RxJs - преобразовать массив Observables в массив выданных значений

Question

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

1. фильтр для действительного (не ноль)массивы cron epressions
2. сопоставляют каждое выражение cron с вызовом службы

---

this.formGroup.valueChanges.pipe(
    op.filter(v => !!v.cronExpressions),
    op.map((v): string[] => v.cronExpressions),
    op.map((v: string[]) => v.map(cron =>
            this.cronService.getReadableForm(cron).pipe(
                op.map(this.toDescription),
                op.map((description): CronExpressionModel => ({ cron, description }))
            )
        )
    ),
    // What now?
).subscribe((cronExpressions: CronExpressionModel[]) => ...) // Expected result

Я хотел бы получить на subscribe() массивCronExpressionModel вернулся со всех вызовов службы.

Я не могу обернуться вокруг этого.

---

Текущее решение, согласно ответу Мартина:

filter(v => !!v.cronExpressions),
map(v => v.cronExpressions),
map(cronExprs => cronExprs.map(c => this.invokeCronService(c))),
mergeMap(serviceCalls => forkJoin(serviceCalls).pipe(defaultIfEmpty([])))

Answer 1

Чтобы преобразовать поток в массив, вы можете использовать оператор toArray .

Вот предложение:

this.formGroup.valueChanges.pipe(
    filter(v => !!v.cronExpressions),
    // transform [item1, item2...] into a stream ----item1----item2----> 
    concatMap((v): Observable<string> => from(v.cronExpressions).pipe(
        // for each of the items, make a request and wait for it to respond
        concatMap((cron: string) => this.cronService.getReadableForm(cron)),
        map(this.toDescription),
        map((description): CronExpressionModel => ({ cron, description })),
        // wait for observables to complete. When all the requests are made, 
        // return an array containing all responses
        toArray()
      )
    ).subscribe((cronExpressions: CronExpressions[]) => ...) // Expected result

Примечание:

Вы можете использовать mergeMap вместо concatMap для распараллеливания запросов.Но вам нужно знать, что вы делаете;)

Answer 2

Вы можете просто добавить forkJoin, если не возражаете против параллельного выполнения всех запросов:

switchMap(observables => forkJoin(...observables))

или если вы хотите выполнить все из них в последовательности:

switchMap(observables => concat(...observables).pipe(toArray()))

Вместо switchMap вы можете использовать concatMap или mergeMap в зависимости от того, какое поведение вы хотите.