Combine Observables into a single Observable from a dynamic array of Observable?

Question

Summary: essentially I need something like from Users where usergroupIds in [1,3,5].

Given usergroupIds$ emit an array of group IDs [1,3,5]

I want to union all users by userids in usergroupIds and combine the userIds (distinct)

Here is what I came up with:

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

Unsubscribe & resubscribe users$ every time on change seems wrong?

Is it possible to express users$ fully in RxJS without creating it in a subscription every time?

Update: With help from @FanCheung:

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

However, since usersObservable emit an array of users at a time, The combinedUsers$ results in somthing like [[userA, userB], [userB, userC], [userA, userD]] which I don't mind to carry out the extra processing on subscribe:

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

However, is there a way to use somehow flatten the results of combinedUsers$, and then carry out the distinct operator?

Fan Cheung · Accepted Answer · 2019-01-16T02:34:46.420

1

See if this work. Basically it fires when usergroupIds$ emits and thus you get your new dynamic observable. shareReplay is needed if you always want the source observable usergroupdIds$ to emit the last saved value when subscribed

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

edited Jan 16 '19 at 02:34

answered Jan 15 '19 at 02:05

Fan Cheung

10,745
3
17
39

Tried this but i get [[usersInGroup1, usersInGroup2, usersInGroup3]]. Help? Thanks – Henry Jan 15 '19 at 19:02
@Henry Changed combinelatest to forkjoin and these three line should make filtering work `map(arr=>[].concat(...arr)), switchMap(arr=>from(arr)), distinct(user => user.id)` – Fan Cheung Jan 16 '19 at 02:35
Can you pls explain what does the `map(arr=>[].concat(...arr))` line do? Why is it copying the array into an empty array, wouldn't that still make 3 separate arrays? – Henry Jan 16 '19 at 04:53
1

concat takes multiple array as argument, you start with an empty array [] and `...arr` basically explode the arrays into arguments. your array will become [].concat(arr1,arr2,arr3) and flattened https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Array/concat – Fan Cheung Jan 16 '19 at 05:46

score 0 · Answer 2 · answered Jan 15 '19 at 10:36

0

So you don't want to be making separate requests to createUsersObservable every time usergroupIds$ emits. This sounds like a good usecase for mergeScan:

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

Live demo: https://stackblitz.com/edit/rxjs-8zybdu

answered Jan 15 '19 at 10:36

martin

93,354
25
191
226

Performant in terms of what? It makes just one `createUsersObservable()` call per emission from `usergroupIds$` so in that case yes. – martin Jan 15 '19 at 10:44

Combine Observables into a single Observable from a dynamic array of Observable?

2 Answers2