Flattening Nested Observables

Question

I have the following block of code for my nested observable and am trying to get observable<user[]> which this code is not doing:

return this.findUser(term).map( users => {
  return users.map( user => this.getLastLogin(user.user_id).map( last_login => {
    user.last_login = last_login;
    return user;
  }));
});

findUser returns Observable<User[]> and getLastLogin returns Observable<number>.

I'm basically hoping to fetch a list of users and then update this with the information from another value.

Right now the code above is returning <Observable<Observable<User>[]>.

I thought I could replace the initial map with flatMap but this turns the object into <Observable<Observable<User>>.

The RxJS documentation is a little hard to decipher so I'm not sure what combination of switch, forkJoin or flatMap will get me to what I need.

How can I change the code to just return an Observable<User[]>?

Drayman

Answer 1

Actually, you don't need forkJoin() nor switch() to do this.

In general, you want to update each user in the array of users by another async call.

I'd do it like this:

var source = findUser('term')
    .mergeAll()
    .mergeMap(user => getLastLogin(user.user_id)
        .map(last_login => {
            user.last_login = last_login;
            return user;
        })
    )
    .toArray();

source.subscribe(val => console.log(val));

Operator mergeAll() converts a higher-order Observable into single observables. In this case it takes the array of all users and re-emits them one by one. Then mergeMap() emits users updated with the last_login date. At the end I used toArray() to transform single users into one large array that is them emitted as whole (you can remove this operator if you want to emit single users instead).

Note that when you used return users.map(...) you were using Array.map() that returns an array and not map() from RxJS that returns an Observable. I think working with single objects is usually easier that with arrays of objects.

See live demo: https://jsbin.com/naqudun/edit?js,console

This prints to console:

[ { name: 'foo',
    user_id: 42,
    last_login: 2016-11-06T10:28:29.314Z },
  { name: 'bar',
    user_id: 21,
    last_login: 2016-11-06T10:28:29.316Z } ]

Reptant

I think yours is the neater answer - especially as the question's title mentions flattening nested observables - but I'm curious as to whether the array order would always match that of the source array: would the use of mergeMap result in a non-deterministic order? Would the order depend upon which to-be-merged observables emit first? – Dowson

This is true, either merge nor mergeAll guarantees the same order which might or might not be an issue. However, if it is an issue you could just replace mergeMap with concatMap and that's it. – Reptant

Interestingly this code doesn't work when using TypeScript's classes. When compiled it looks like mergeAll on the findUsers method is returning a User[]. Here's a link to the behaviour I'm seeing. jsbin.com/wiwayikiye/edit?js,console Any ideas? – Drayman

@Drayman I've never seen syntax you're using to create User class. I don't even think that's a valid TypeScript syntax. See this jsbin.com/gikabu/2/edit?js,console – Reptant

Doh. Weirdly enough it's working perfectly. There's errors being thrown saying "Property 'mergeMap' does not exist on type 'User[]'". It could be some weird angular/typeScript business but who knows? Any ideas? – Drayman

Answer 2

One solution would be to use forkJoin to join and then map the results of calls to getLastLogin to users:

// forkJoin will return Observable<User[]>, so use switchMap.

return this.findUser(term).switchMap(users => Observable.forkJoin(

  // Map the array of users to the array of observables to be joined. Use
  // first to ensure the observables complete.

  users.map(user => this.getLastLogin(user.user_id).first()),

  // Use forkJoin's selector to add the last_login to each user and return
  // the users.

  (...logins) => {
    users.forEach((user, index) => { user.last_login = logins[index]; });
    return users;
  }
));

Dowson

Answer 3

As @martin pointed out, mergeAll() is the solution. You can also apply it at the end of your map operations.

With more recent syntax, this reads:

getStations(factoryId: number): Observable<Station[]> {//...omitted}

getAllStations(): Observable<Station[]> {
const result =  this.factoriesService.getFactories().pipe(
  map(factories => factories.map(f => f.id)),
  mergeMap(factoryIds => factoryIds.map(id => this.getStations(id))),
  mergeAll()
);

return result;

}

Pedroza