Есть ли оператор, который работает как concatMap, но с несколькими внутренними наблюдаемыми

Question

Я использую наблюдаемое для запроса моей БД. Эта наблюдаемая вернет массив со всеми найденными соответствующими объектами. Моя проблема в том, что я хочу отобразить наблюдаемое с более подробной информацией, которую я получу из другого API.

Я попробовал concatMap, но он позволил мне только вложить 1 наблюдаемую в исходную наблюдаемую

const usersAPI$: Observable<Users[]> = //  something
const assistsAPI$: Observable<Assists[]> = //  something

const modifiedAssists$ = assistsAPI$.find().pipe(
  concatMap(assists =>
    assists.map(assist => usersAPI$.get(assist.userID)
       .pipe(
          map(user => 
            assist = {...assist}, ...{userName: user.userName}
          )
        )
    )
  )
);

Вы можете увидеть похожий рабочий пример здесь, на stackblitz, https://stackblitz.com/edit/rxjs-concatmap-issue-stackoverflow где «результат» является правильным способом использования concatMap и «result2» - это нерабочий способ, который я ожидал сработать

Answer 1

Вы должны каким-то образом обработать ваш внутренний массив Observables.Вы можете использовать forkJoin, merge или concat в зависимости от ваших потребностей.

forkJoin

forkJoin будет обрабатывать ваши внутренние вызовы API параллельно и возвращать массив, когда все APIзвонки завершены.Обратите внимание, что внутренние Observables должны завершиться.Я думаю, это то, что вам должно подойти в вашем приложении.

import { forkJoin } from 'rxjs';

const modifiedAssists$ = assistsAPI$.find().pipe(
  concatMap(assists => forkJoin(assists.map(assist => someObservable ))
);

merge

merge подпишется на все внутренние вызовы API сразу и отправит результаты всех вызовов один за другимпо мере их поступления.

import { merge } from 'rxjs';

const modifiedAssists$ = assistsAPI$.find().pipe(
  concatMap(assists => merge(...assists.map(assist => someObservable )),
  // toArray() add the toArray operator if you need the result as an array
);

Это даст вам желаемый результат в вашем стеке блиц, но я думаю, что ваш стек бликов несколько ошибочен, а не то, что вы ищете в примере кода из вашего вопроса, так какinner Observable в вашем стеке выделяет несколько раз, и окончательный результат не является массивом.Если порядок внутренних запросов не имеет значения, все они выдают одно значение, а затем завершаются, и вам нужны результаты всех запросов в виде массива, просто используйте forkJoin.

concat

concat будет подписываться на все внутренние вызовы API один за другим.Следующая наблюдаемая в последовательности будет подписана только после завершения предыдущей.Поэтому выполнение будет медленнее, чем с forkJoin или merge, поскольку следующий HTTP-запрос будет выполнен только после того, как предыдущий вернул значение.Используйте это, если вызовы вашего userAPI должны выполняться в том же порядке, что и assists в вашем массиве.

import { concat } from 'rxjs';

const modifiedAssists$ = assistsAPI$.find().pipe(
  concatMap(assists => concat(...assists.map(assist => someObservable )),
  // toArray() add the toArray operator if you need the result as an array
);

Answer 2

Я полагаю, что вы хотите mergeMap?

Он отображает внешнее наблюдаемое на разрешенное внутреннее, наблюдаемое для каждого внешнего излучения.

const result = clicks.pipe(
  concatMap(ev =>
    interval(1000).pipe(take(4))
  )
);

const result2 = clicks.pipe(
  mergeMap(ev =>
    arr.map(() => interval(1000).pipe(take(4)))
  )
);
result.subscribe(x => {
  console.log(x)
});

result2.subscribe(x => {
  console.log(x)
});

Answer 3

Вы можете использовать combineLatest() (как статический импорт без конвейера)

import {combineLatest} from 'rxjs';

const usersAPI$: Observable<Users[]> = //  something
const assistsAPI$: Observable<Assists[]> = //  something

const combinedObservable$ = combineLatest(usersAPI$,assistsAPI$, someOtherStuff$);

const modifiedAssists$ = combinedObservable$.pipe(
   map([usersApiValues, assistsAPIValues, someOtherStuffValues] => { 
      /** what you want here */
   }) 
);

Просто учтите, что каждый из наблюдаемых внутри combineLatest должен излучать хотя бы один раз (например, вы можете начать их с пустого массива)