javascript - RxJS: JSON data with an array, processing each item further in the stream

Question

I receive a HTTP response, which contains, if all goes well, a single array, coded as JSON.
I want to get this array, filter out some items, and process the passing items as events.

What I do so far is:

    return this._http.get(url)
        .map((res:Response) => res.json())
        .map((data:any) => {
            if (!Array.isArray(data) || data.length == 0) {
                throw new Error("No items returned, URL: " + url);
            }
            let projects = <ProjectModel[]>service.fromJSONarray(data, this._http);
            return Observable.from(projects)
                .filter((project: ProjectModel) => project.parentProject == null)
                .subscribe(project -> ...)
        })

However I don't like the nesting. I assume there's a way to do this:

    return this._http.get(url)
        .map((res:Response) => res.json())
        .map((data:any) => {
            ...
            let projects = <ProjectModel[]>service.fromJSONarray(data, this._http);
            ???
        })
        .filter((project: ProjectModel) => project.parentProject == null)
        .subscribe(project -> ...)

How to achieve that?

Answer 1

Assuming you're using RxJS 5 there's even easier way with concatAll() and its undocumented feature for flattening nested arrays even though concatAll() is made to work with Higher-Order Observables.

let data = '[{"id":1},{"id":2},{"id":3},{"id":4}]';

Observable.of(data)
    .map(JSON.parse)
    .concatAll() // each object is emitted separately
    .map(p => p) // transform to ProjectModel
    .filter(p => true) // filter whatever you want
    .subscribe(v => console.log(v));

If service.fromJSONarray returns just a single instance of ProjectModel it can be this simple. If you need service.fromJSONarray to return an array you'd put it between map(JSON.parse) and concatAll(). This demo prints to console:

{ id: 1 }
{ id: 2 }
{ id: 3 }
{ id: 4 }

However, if you were returning an Observable from service.fromJSONarray that emits other observables because it needs to do some extra HTTP requests you'd need to use concatMap() or flatMap() depending on whether you want to keep the same order of items or not:

let data = '[{"id":1},{"id":2},{"id":3},{"id":4}]';

function fromJSONarray(arr) {
    return Observable.from(arr)
        .flatMap(p => Observable.of(p));
}

Observable.of(data)
    .map(d => fromJSONarray(JSON.parse(d)))
    .concatAll()
    .map(p => p) // transform to ProjectModel
    .subscribe(v => console.log(v));

The result is the same.

Similar questions:

• Subscribing to a nested Observable

• Merge subarrays using Observables

• RxJS Refactor nested map statement