Объединить наблюдаемые в один наблюдаемый из динамического массива наблюдаемых?

Question

Резюме: по сути, мне нужно что-то вроде from Users where usergroupIds in [1,3,5].

Дано usergroupIds$ создать массив идентификаторов групп [1,3,5]

Я хочу объединить всех пользователей по userids в usergroupIds и объединить идентификаторы пользователей (разные)

Вот что я придумал:

usergroupIds$.subscribe(usergroupIds => {
    const users$s = usergroupIds.map(gid => createUsersObservable(gid))
    //  [users$ in group 1, users$ in group 3, users$ in group 5]
    const users$ = combineLatest(...user$s).pipe(
        distinct(user => user.id)
    )
})

createUsersObservable = gid => 
  collectionData(db.collection('users').where('groupId', '==', gid)) // rxFire firestore

Отмена подписки и повторная подписка users$ каждый раз, когда изменения кажутся неправильными?

Можно ли выразить users$ полностью в RxJS, не создавая его каждый раз в подписке?

Обновление: С помощью @FanCheung:

combinedUsers$ = userGroupIds$.pipe(
        switchMap(userGroupIds => {
            const users$s = userGroupIds.map(groupId =>
                createUsersObservable(groupId))
            return combineLatest(...users$s)
        })

Однако, так как usersObservable генерирует массив пользователей за один раз, комбинированный пользователь $ приводит к чему-то вроде [[userA, userB], [userB, userC], [userA, userD]], что я не против выполнить дополнительную обработку при подписке:

combinedUsers$.subscribe(combinedUsers => {
        const userMap = {}
        for (const users of combinedUsers) 
            users.forEach(user => (userMap[user.id] = user))
        const uniqueUsers = Object.values(userMap)

        // update UI to uniqueUsers
    })

Однако есть ли способ как-то сгладить результаты combinedUsers$, а затем выполнить оператор distinct?

Answer 1

Так что вы не хотите делать отдельные запросы к createUsersObservable каждый раз, когда usergroupIds$ отправляет. Это звучит как хороший пример использования mergeScan:

usergroupIds$.pipe(
  mergeScan((users, user) => createUsersObservable().pipe(
    map(user => [user, ...users]),
  ), []),
);

Демонстрация в реальном времени: https://stackblitz.com/edit/rxjs-8zybdu

Answer 2

Посмотри, работает ли это. По сути, он срабатывает, когда usergroupIds $ испускает, и, таким образом, вы получаете новую динамическую наблюдаемость. shareReplay необходим, если вы всегда хотите, чтобы наблюдаемый источник usergroupdIds$ испускал последнее сохраненное значение при подписке

usergroupIds$.pipe(
shareReplay(1),
switchMap(usergroupIds => {
    const users$Array = usergroupIds.map(gid => createUsersObservable(gid))
    //  [users$ in group 1, users$ in group 3, users$ in group 5]
   return forkJoin(...users$Array).pipe(
        map(arr=>[].concat(...arr)),
        switchMap(arr=>from(arr)),
        distinct(user => user.id)
    )
})
)