如何使用rxjs expand强制执行深度优先检索？

Question

我正在研究Angular 7 / Typescript / RxJS 6.3.3项目，使用各种RxJS Observable和相关运算符来处理通过http服务器从数据库检索的对象的分层集合（特定于树状）。

我想用expand算子创建深度优先搜索，并使用concatMap来保持秩序...或者我希望如此。不行。

请参阅此处的蒸馏示例：https://stackblitz.com/edit/rxjs-vf4zem

该示例的控制台输出是：

dfs no delay: [1,8,9,22,23,24,2,4,7,3]
dfs with delay: [1,2,3,4,7,8,9,22,23,24]

（第二个输出行可能会根据添加的延迟量而有所不同。延迟旨在模拟从http服务器获取数据。）

鉴于示例中的数据，我的愿望是始终获得第一行输出：深度优先排序。该示例的关键功能：

const dfs = (getter: Getter) => rootIds.pipe(
  concatMap(ids => from(ids)),
  expand(id =>
    getter(id).pipe(
      concatMap(children => from(children))
    )
  ),
  toArray()
);

有没有办法强制执行深度优先处理？ expand可以不保证，或者这只是一个不好的方法来完成将分层数据变成一个扁平的深度优先阵列？

Answer 1

我认为这是一个很好的问题，我同意，对于并行获取，你需要一些额外的数据结构来获取后的结果。

然而，通过expand实现递归重建很有意思，所以继承我的顺序尝试：

sequentialDFS(getChildren: Getter, ids: number[]): Observable<number[]> {
  return of(ids).pipe(
    expand(([head, ...rest]) =>
      // here we have a sequence of ids
      // that we'll explore in left-to-right order,
      // e.g. [1, 17, 20] will...
      getChildren(head).pipe(
        switchMap(subChildren => {
          // ...will turn here into [2, 6, 17, 20]
          const acc = [...subChildren, ...rest ];
          return acc.length
            ? of(acc)
            : EMPTY;
        })
      )
    ),

    // collect the heads {{{
    map(([head])=>head),
    toArray()
    // }}}
  );
}

*我有点修改了getChildren方法来返回Observable<number[]>而不是Observable<number>

https://stackblitz.com/edit/rxjs-rf8d1j

这绝不是并行获取的答案。只是分享它，因为它很有趣。

Answer 2

所以我玩了一下，做了一些手动递归（而不是依靠expand），并提出了以下内容（也在上面的stackblitz链接更新）：

class Test {
  // This doesn't maintain depth-first order; it can vary depending on timing.
  brokenDFS(getChildren: Getter, ids: number[]): Observable<number[]> {
    return of(ids).pipe(
      concatMap(ids => from(ids)),
      expand(id => getChildren(id)),
      toArray()
    );
  }

  workingDFS(getChildren: Getter, ids: number[]): Observable<number[]> {
    return from(ids).pipe(
      concatMap(id => this.parentAndChildren(getChildren, id)),
      toArray()
    );
  }

  private parentAndChildren(getChildren: Getter, id: number): Observable<number> {
    return of(id).pipe(
      concat(
        getChildren(id).pipe(
          map(child => this.parentAndChildren(getChildren, child)),
          concatAll()
        )
      ),
    );
  }

}

const getter = getChildrenWithDelay;
const rootIds = [1, 17, 20];
const test = new Test();
test.brokenDFS(getter, rootIds).subscribe(data => console.log(`Broken: ${data}`));
test.workingDFS(getter, rootIds).subscribe(data => console.log(`Working: ${data}`));

输出（注意“Broken”输出可能因时序而变化）：

Broken: 1,17,20,21,2,6,18,7,13,14,3,4,19,15,16,5,8,9,10,11,12
Working: 1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21

（另请注意，我从原始帖子更改了树值，使得正确的DFS是从1到21的数字数组）。

所以workingDFS工作，但比brokenDFS要慢得多，因为每个对http服务器的请求必须等待它之前的所有内容才能完成，而brokenDFS版本会同时运行多个请求（不按正确的顺序）。

我不知道这里是否有更好的rxjs选项。我可能不得不修改我的方法，不仅传递感兴趣的对象，还传递一些结构/排序信息，同时发出所有/许多请求，然后以正确的顺序组合所有请求。