Angular 2中基于RxJS实现N叉树递归上传并监听完成状态

当然可以！RxJS的高阶Observable操作符正好能完美解决这个问题——我们可以把递归上传的逻辑包装成一个统一的Observable，它只会在整棵树的所有节点都上传完成后发射一次结果。

核心思路 #

原来的代码直接订阅每个节点的上传请求，没有统一的完成信号。我们需要：

1. 先上传当前节点，拿到它的NodeJson（包含子节点需要的父ID）。
2. 递归处理所有子节点的上传，等待所有子节点的上传请求都完成。
3. 最后返回一个Observable，只有当当前节点+所有子节点都上传成功时，才发射当前节点的NodeJson。

实现代码 #

我们可以用concatMap（保证当前节点上传完成后再处理子节点）和forkJoin（等待所有子节点上传完成）来实现：

import { Observable, of, forkJoin } from 'rxjs';
import { concatMap, map } from 'rxjs/operators';

// 你的目标方法实现
uploadNodeTree(node: TreeNode, parentId: number): Observable<NodeJson> {
  // 第一步：上传当前节点
  return this.node_service.addNode(node, parentId).pipe(
    // 上传成功后，处理子节点
    concatMap(uploadedNode => {
      // 如果没有子节点，直接返回当前节点的Observable
      if (!node.children || node.children.length === 0) {
        return of(uploadedNode);
      }

      // 第二步：为每个子节点创建递归上传的Observable
      const childUploadObservables = node.children.map(child => 
        this.uploadNodeTree(child, uploadedNode.id)
      );

      // 第三步：等待所有子节点上传完成，然后返回当前节点的结果
      return forkJoin(childUploadObservables).pipe(
        // 忽略子节点的返回结果，只发射当前节点的NodeJson
        map(() => uploadedNode)
      );
    })
  );
}

使用方式 #

现在你可以订阅这个方法返回的Observable，当它发射值时，就代表整棵树已经完全上传完成：

// 假设rootNode是你的N叉树根节点，parentId是根节点的父ID（比如0或者业务约定的根父ID）
this.uploadNodeTree(rootNode, parentId).subscribe({
  next: (rootUploadedNode) => {
    console.log('整棵树上传成功！根节点ID:', rootUploadedNode.id);
    // 这里可以添加上传完成后的逻辑，比如提示用户、刷新数据等
  },
  error: (error) => {
    console.error('上传过程中出现错误:', error);
    // 处理错误，比如回滚操作、提示用户上传失败等
  }
});

可选的错误处理 #

如果希望某个子节点上传失败时，不中断整个树的上传（根据业务需求决定），可以在递归调用时添加catchError操作符：

import { catchError } from 'rxjs/operators';

// 修改子节点Observable的创建逻辑
const childUploadObservables = node.children.map(child => 
  this.uploadNodeTree(child, uploadedNode.id).pipe(
    catchError(err => {
      console.error(`子节点上传失败:`, err);
      return of(null); // 返回一个占位值，让forkJoin继续等待其他节点完成
    })
  )
);

关键操作符说明 #

• concatMap：确保当前节点的上传请求完成后，才开始处理子节点的上传，保证了节点上传的顺序（如果不需要严格顺序，也可以用mergeMap实现并行上传）。
• forkJoin：会等待传入的所有Observable都完成后，才发射一个包含所有结果的数组。这里我们用它来等待所有子节点的上传请求结束。

内容的提问来源于stack exchange，提问作者Mathieu Mandret