使用 Observable 进行递归答案

【问题标题】：Recursion with Observables使用 Observable 进行递归
【发布时间】：2018-11-02 03:29:20
【问题描述】：

我需要在我的 Angular 2 项目中使用 REST API 上传 n 元树。我有一项服务允许我添加这样的节点：

addNode(node: TreeNode, parentId: number): Observable<NodeJson>

TreeNode 是一个对象：

{
  data: any,
  children: TreeNode[]
}

我需要父节点的id才能添加子节点，这个id在NodeJson对象中。

我目前的解决方案是：

uploadTreeNode(node: TreeNode, parentId: number) {
    this.node_service.addNode(node, parentId)
        .subscribe(uploaded_node => {
            for (let child of node.children) {
                this.uploadTreeNode(child, uploaded_node.id)
            }
        }

}

这很好，但这是我的问题：我需要知道整棵树何时已上传，但使用这种方法是不可能的，因为我在每次调用时都直接订阅 observable，而且我无法知道它何时完成。

实际上，我想要实现的是拥有这个功能：

uploadNodeTree(node: TreeNode, parentId: number): Observable<NodeJson>

它会返回一个 Observable，它会在子上传的所有 Observable 完成后发出。

有没有办法用 Rxjs 操作符做这样的事情？

【问题讨论】：

您可以使用forkJoin 了解所有可观察对象何时发出。
是的，但是 forkJoin 将 Observables 列表作为参数，但我不知道如何构建此列表，因为我不跟踪我订阅的 Observables。
您可能想使用expand 运算符。因此，只要您的 observable 返回带有父节点的节点，您就可以使用 expand 运算符再次调用整个函数。参考：learnrxjs.io/operators/transformation/expand.html
我不认为你想订阅一个开始，因为你的函数将返回一个订阅而不是一个 Observable。
@deezg expand 似乎是一个很好的解决方案，但正如你所说，我需要一个返回每个节点及其父 ID 的函数，这就是我被困的地方，因为我有几个孩子在每个节点上我都找不到让这样一个功能起作用的方法

标签： angular typescript recursion rxjs

【解决方案1】：

正如所指出的，expand 是正确的运算符。解决方案是上传第一个节点并使用expand 创建来自其子节点的请求流并将其“附加”到源流中，一旦没有更多子节点，只需发送 empty() 以结束流：

uploadTreeNode(node: TreeNode): Observabe<NodeJson> {
    return this.node_service.addNode(node, rootId)
        .map(uploaded_node => ({id: uploaded_node.id, current_node: node})
        .expand({id, current_node} => {
            if (current_node && current_node.children) {
                let reqs = current_node.children.map(n => this.nodeService.addNode(id, n))
                return Observable.merge(...reqs)
            } else {
                return Observable.empty()
            }
        })   
}

现在我可以做：

this.uploadTreeNode(myNode, 1).subscribe(uploaded_node => {
    console.log('Node has been uploaded: ' + uploaded_node),
})

感谢大家的帮助。

【讨论】：