RxJs Array of Observable to Observable of Array

Question

For a Web-Application written with Angular2 in TypeScript, I need to work with RxJs Observables.
As I never used rxjs before and I am new to reactiveprogramming in general, I sometimes have some difficulties finding the right way to do some specific things.
I am now facing a problem. where I have to convert an Array<Observable<T>> into an Observable<Array<T>>.
I'll try to explain it with an example:

I have an Observable, which gives me a list of Users (Observable<Array<User>>)
The User-class has a function getPosts returning an Observable<Array<Post>>.
I need to map the Observable<Array<User>> to an Observable<Array<Post>> to evaluate all the Posts inside the onNext function.

I can easily map from Observable<Array<User>> to Observable<Array<Observable<Array<Post>>>> using
map((result : Array<User>) => result.map((user : User) => user.getPosts()))
and I can flatten an Array<Array<Post>> into an Array<Post>.
However I just can't find the correct way to map the Observable<Array<Observable<Array<Post>>>> into an Observable<Array<Array<Post>>>
Until now i used the combineLatest function together with flatMap.
To me it seemed to work and the editor i used (Atom editor) did not show any error. However now I use Netbeans, which shows me an error in this code. Also compiling the code using "tsc" results in the following errors:

Argument of type '(result: Post[]) => void' is not assignable to parameter of type 'NextObserver<[Observable<Post>]> | ErrorObserver<[Observable<Post>]> | CompletionObserver<[...'.
Type '(result: Post[]) => void' is not assignable to type '(value: [Observable<Post>]) => void'.

So my question is:
How can I "flatten" an Array of Observables into an Array?

What does the getPosts method correspond to? Does it load the posts of a specific user? — Thierry Templier
– Thierry Templier, Commented May 2, 2016 at 14:07

Thierry Templier · Accepted Answer · 2016-05-02 14:11:39Z

41

The flatMap operator allows to do that. I don't fully understand what you try to do but I'll try to provide an answer...

If you want load all the

getPostsPerUser() {
  return this.http.get('/users')
    .map(res => res.json())
    .flatMap((result : Array<User>) => {
      return Observable.forkJoin(
        result.map((user : User) => user.getPosts());
    });
}

Observable.forkJoin allows you to wait for all observables to have received data.

The code above assumes that user.getPosts() returns an observable...

With this, you will receive an array of array of posts:

this.getPostsPerUser().subscribe(result => {
  var postsUser1 = result[0];
  var postsUser2 = result[1];
  (...)
});

answered May 2, 2016 at 14:11

Thierry Templier

203k44 gold badges407 silver badges365 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

Robert P Over a year ago

Yea the getPosts() returns an Observable<Array<Post>>. So I guess thats what I am looking for. I'll try it out asap and let you know if it worked. Thanks

Robert P Over a year ago

Works perfectly. I only hat to replace combineLatest with forkJoin. Thanks a lot!

izogfif · Accepted Answer · 2018-05-18 17:48:37Z

10

You can use the function apply on the rxjs method you want, like this:

const source1 = Rx.Observable.interval(100)
  .map(function (i) { return 'First: ' + i; });

const source2 = Rx.Observable.interval(150)
  .map(function (i) { return 'Second: ' + i; });

const observablesArray = [source1, source2];

const sources = Rx.Observable.combineLatest
.apply(this, observablesArray).take(4)

/* or you can write it without "apply" this way:
const sources = Rx.Observable
                .combineLatest(...observablesArray)
                .take(4)
*/
sources.subscribe(
  (response) => {
    console.log(response);
  }
)

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.1/Rx.min.js"></script>

edited May 18, 2018 at 17:48

izogfif

7,9252 gold badges39 silver badges34 bronze badges

answered Jan 12, 2017 at 9:44

Lievno

1,0419 silver badges17 bronze badges

1 Comment

Robert P Over a year ago

I guess this should work to, but I'll stick with forkJoin for now.

Paul John Leonard · Accepted Answer · 2018-10-06 16:55:08Z

5

Here is some example code using [email protected]

import { of, forkJoin} from 'rxjs';
import { Observable } from 'rxjs'

const source:Array<Observable<String>> = [of('A'), of('B'), of('C') ];

forkJoin(source).subscribe( (x)=> console.log(x))

https://arrayofobservablesexamp.stackblitz.io

answered Oct 6, 2018 at 16:55

Paul John Leonard

4384 silver badges7 bronze badges

Comments

Adventitious Angles Qs Poster · Accepted Answer · 2018-05-24 12:46:01Z

4

Use forkJoin on this array of observables, then it will be an observable of array.

answered May 24, 2018 at 12:46

Adventitious Angles Qs Poster

7218 silver badges22 bronze badges

Comments

FindOutIslamNow · Accepted Answer · 2019-10-27 14:53:00Z

0

Using toArray()

range(0, 7).pipe(
  concatMap(v => of(`day_${v}`)),
  toArray() // here
).subscribe(a => {
  console.log(a); // ['day_0', 'day_1', ... 'day_6']
});

answered Oct 27, 2019 at 14:53

FindOutIslamNow

1,2661 gold badge16 silver badges37 bronze badges

Collectives™ on Stack Overflow

RxJs Array of Observable to Observable of Array

5 Answers 5

2 Comments

1 Comment

Comments

Comments

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

5 Answers 5

2 Comments

1 Comment

Comments

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related