Observable of array for array of Observables

Question

I have the following situation using Observables in my Angular4 app that I just can't get to work: I want to gather summary data of all my booking days for a overview page. Getting all the days is an Observable, and each day has a list of bookings of that day that I have to retrieve - again an observable source. From this list I calculate a summary of the day. All these summaries I want to emit in a resulting observable.

I have tried lot's of more complicated things, but always the inner observables where not waited on to complete and I got empty summaries. I have gotten back to the basics, and something along these lines should work:

getSummaries(): Observable<BookingDaySummary[]> {
    return this.bookingService.getBookingDays().take(1).mergeMap(
        days => this.calculateSummaryOfDays(days)
    )
};

private calculateSummaryOfDays(days: BookingDay[]): Observable<BookingDaySummary[]> {
    const summaries$ = days.map(day => this.calculateSummary(day));
    // I'm aware that the next line is not correct. 
    // Essentially I want the array of observables to complete 
    // and have an array of the resulting values.
    return Observable.merge(summaries$);
}

private calculateSummary(day: BookingDay): Observable<BookingDaySummary> {
    // ... logic to get summary from one day
}

However, the type of summaries$ is Observable<Observable<BookingDaySummary> and not Observable. So it all boils down to: How do I make an Observable<T[]> from [Observable<T>]?

Also: Should the most inner method I use in .map return an Observable or just be a map on the incoming type to T when intending to produce an Observable<T>?

Velocity

Does calculateSummary really (need to) return an Observable? – Finecut

@IngoBürk: That was part of the question. I have tried both. The method itself uses an Observable (another db fetch), so I guess it has to. – Velocity

Answer 1

How do I make an Observable< T[] > from Observable< T >[] ?

forkJoin does this. Using String for your T:

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

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

const source2:Observable<String[]> = forkJoin(source);

source2.subscribe((res=>console.log(res)));

https://stackblitz.com/edit/arrayofobs2obsofarray?file=index.ts

Unstoppable

Answer 2

Use the RxJS combineLatest static function on your last statement of calculateSummaryOfDays:

return Observable.combineLatest(summaries$);

Note that your summaries$ is an array of observables:

const summaries$: Observable<BookingDaySummary>[] = ...

From the combineLatest documentation:

Combines multiple Observables to create an Observable whose values are calculated from the latest values of each of its input Observables.

Static version of combineLatest accepts either an array of Observables or each Observable can be put directly as an argument.

Thus this function will take your array of observables and return an observable of an array.

Fishing

As a comment to this answer: "if you are working with observables that only emit one value, or you only require the last value of each before completion, forkJoin is likely a better option." (quoting straight from the documentation here) – Linzy

Answer 3

try

return Observable.from(summaries$).mergeMap(value=>value);

I suppose summaries is an array of observable - Observable[] use Observable.from will emit them one by one and followed by mergeMap to flatten and execute the Observable and in the end you will get plain value emission.

Vicious

Observable<BookingDaySummary> is not assignable to type Observable<BookingdaySummary[]> happens when I try your suggestion. – Velocity

change your typing from Observable<BookingDaySummary[]> to Observable<BookingDaySummary> or just use <any> to see if it works first – Vicious

While this code snippet may solve the question, including an explanation really helps to improve the quality of your post. Remember that you are answering the question for readers in the future, and those people might not know the reasons for your code suggestion. – Linkwork

Answer 4

Here's my solution:

Vector<Observable<Integer>> vs = new  Vector<Observable<Integer>>();
vs.add(Observable.just(1));
vs.add(Observable.just(2));
vs.add(Observable.just(3));
vs.add(Observable.just(4));
vs.add(Observable.just(5));

Vector<Integer> v = new Vector<Integer>();
Single<Vector<Integer>> o = Observable.merge(vs)
              .collectInto(v, Vector<Integer>::add);

Ripsaw

*merge converts an array of observables into a single observable, and collectInto takes all the events an observable emits and combines them into a single observable with a collection such as a vector inside. – Ripsaw