RXJS 递归展开一棵树

Question

我有一个节点树，每次用户“扩展”一个节点时，都会调用一个 http 请求来获取它的子节点。

我正在寻找一个 RXJS 管道来递归地扩展所有树节点并发出扩展的树。

(StackBlitz demo)

我现在的做法是对源进行变异，并在完成后输出变异的对象。 有没有办法让我简化这个地狱？可能没有变异来源。

// 'ROOT' nodes
const rootNodes = [{ id: 0, parent: true }, { id: 1000, parent: true }];

// map the root nodes to a recursive function that mutate node and returns its children observables
const getChildrenOfRoot$ = rootNodes.map(node => getChildren(node));

// call the observables, in parallels, when done, print the *** mutated *** source.
forkJoin(...getChildrenOfRoot$).subscribe(() => console.log(rootNodes));


function getChildren(node: Node): Observable<Node[]> {

  return getChildrenFromServer(node.id).pipe(

    // if no children returned, don't continue
    filter((children: Node[]) => !!(children?.length)),

    // mutate argument's .children property with the returned children array.
    tap((children: Node[]) => (node.children = children)),

    mergeMap((children: Node[]) => {
      // mape children to observables returning either their children, or, an empty array, based on 'parent' property.
      const getChildrenOfChildren$ = children.map(c => (c.parent ? getChildren(c) : of([])));

      return forkJoin(...getChildrenOfChildren$);
    })
  );
}

Answer 1

TLDR;

StackBlitz app.

---

详细解释

由于我们正在处理递归，我认为如果我们从基本情况开始会更容易。
让我们看看我们最初拥有什么：

const rootNodes = [{ id: 0, parent: true }, { id: 1000, parent: true }];

假设这些是这棵树中唯一的父节点（即他们的子节点将全部都是叶节点），我们的问题就变得更容易了：对于上述数组中的每个节点，我们必须获取其子节点并添加一个新属性到它们的节点，称为children。
为此，让我们创建processNodes 孩子：

function processNodes (nodes: Node[]): Observable<Node[]> {
  return nodes.length 
    ? forkJoin(
        nodes.map(node => node.parent ? processNode(node) : of(node))
      )
    : of([]);
}

请注意，我们也在考虑 nodes 数组 为空的情况，因此，在这种情况下，我们只需返回一个空数组 (of([]))。现在让我们关注这部分：

forkJoin(
  nodes.map(node => node.parent ? processNode(node) : of(node))
)

如果processNodes 的参数是：

[{ id: 0, parent: true }, { id: 1000, parent: true }, { id: 7, parent: false }]

那么预期的输出将是：

[
  { id: 0, parent: true, children: [/* ... */] },
  { id: 1000, parent: true, children: [/* ... */] },
  { id: 7, parent: false }
]

所以，如果一个节点是父节点，那么我们将分离的逻辑封装在processNode函数中：

function processNode (node: Node) {
  return getChildrenFromServer(node.id).pipe(
    mergeMap((children: Node[]) => processNodes(children)),
    map((children: Node[]) => ({ ...node, ...children.length && { children } })),
  );
}

这里是递归发生的地方。给定某个父节点的子节点，我们重复这个过程，最后我们将children 属性添加到父节点。

所以，把所有的部分放在一起：

processNodes(rootNodes).subscribe(console.log)

function processNodes (nodes: Node[]): Observable<Node[]> {
  return nodes.length 
    ? forkJoin(
        nodes.map(node => node.parent ? processNode(node) : of(node))
      )
    : of([]);
}

function processNode (node: Node) {
  return getChildrenFromServer(node.id).pipe(
    mergeMap((children: Node[]) => processNodes(children)),
    map((children: Node[]) => ({ ...node, ...children.length && { children } })),
  );
}