RxJS:带有数组的JSON数据,在流中进一步处理每个项目

Question

我收到一个HTTP响应,如果一切顺利,它包含一个编码为JSON的数组.
我想获取此数组,过滤掉一些项目,并将传递的项目作为事件处理.

到目前为止我做的是:

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

但是我不喜欢筑巢.我假设有一种方法可以做到这一点:

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

怎么实现呢？

Answer 1

假设您正在使用RxJS 5,即使使用高阶Observables,也可以使用更简单的方法concatAll()来展平嵌套数组concatAll().

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

如果service.fromJSONarray只返回它的一个实例ProjectModel就可以这么简单.如果你需要service.fromJSONarray返回一个数组,你可以把它放在map(JSON.parse)和之间concatAll().此演示打印到控制台:

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

但是,如果您从service.fromJSONarray其中返回一个Observable,则会发出其他可观察对象,因为它需要执行一些您需要使用的额外HTTP请求,concatMap()或者flatMap()取决于您是否要保持相同的项目顺序:

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

结果是一样的.

类似的问题:

• 订阅嵌套的Observable

• 使用Observables合并子阵列

• RxJS重构嵌套的映射语句