javascript - 如何递归地合并两个嵌套的 observables

我有一些可观察数组，parents$ 和 children$，定义如下。然后我有一个类 AttachChildren。该类有一个方法 getChild$(parentId)，用于获取指定父项的相关子项。最后，方法 tesCollection() 用于填充父项 parent$ 项的 'children' 属性。我尝试使用 forkJoin 来合并这两个流，以便获得最终数据，但到目前为止还未成功。

以下是相关的数据和类。我在 node.js 中使用 Typescript 和 rxjs 版本 "^7.3.0"。

const parents$ = of([
    {
        parentName: 'a',
        parentId: 0,
        children: []
    },
    {
        parentName: 'b',
        parentId: 1,
        children: []
    },
    {
        parentName: 'c',
        parentId: 2,
        children: []
    },
]);

const children$ = of([
    {
        parentId: 0,
        childName: 'a-a',
        childData: [
            {
                itemId: 0,
                itemName: 'd'
            },
            {
                itemId: 1,
                itemName: 'e'
            }
        ]
    },
    {
        parentId: 2,
        childName: 'a-b',
        childData: [
            {
                itemId: 0,
                itemName: 'f'
            },
            {
                itemId: 1,
                itemName: 'g'
            }
        ]
    }
]);

我需要孩子通过相关的 parentId 属性附加到父母上，该属性在 parent$ 和 children$ 中都是共有的。

我尝试过的是这个：

class AttachChildren {

   tesCollection() {
        parents$
            .pipe(
                mergeMap((m: Parent) => {
                    return m.map(parent => {
                        const child$ = this.getChild$(parent.parentId);
                        return forkJoin(
                            {
                                parent: of(parent),
                                child: child$
                            }
                        )
                    })
                })
            )
            .subscribe((p: any) => {
                console.log('subscribe/parents:', p);
            });

    }

    getChild$(parentId: number): Observable<Child> {
        return children$
            .pipe(
                mergeMap(c => c.map(child => child.parentId === parentId))
            )
    }
}

我得到的结果如下：

subscribe/parents: Observable { _subscribe: [Function] }
subscribe/parents: Observable { _subscribe: [Function] }
subscribe/parents: Observable { _subscribe: [Function] }

我怎样才能把这个结果展平？

答案＃1

我不确定你为什么需要用这么复杂的 RxJS，以一种非常命令式的方式。我更喜欢以一种更函数式的方式来使用它。

这里有一段将父母和子女关联起来的代码片段。即使这并不是你完全想要的答案，我希望它能让你了解代码可以变得多么简单（易于阅读和维护）以及少了多少命令式风格。

forkJoin({
  parents: parents$,
  children: children$,
}).pipe(
  map(({children, parents}) => {
    const getChildrenOfParent = id => {
      return children.filter(child => child.parentId === id)
    }

    return parents.map(parent => ({
        ...parent,
        children: getChildrenOfParent(parent.parentId)
    }));
  })
).subscribe(console.log)

输出是：

我在这里使用 forkJoin，因为这是一个 Observable 创建器，当所有流都完成时它会发出数据——对于像本例中组合多个 of 来说完全没问题。在实际应用中，如果你预计父流和子流会在随机时间到达，你也可以使用 combineLatest。一旦每个流至少发出一次数据，它就会开始发出数据。