重构链式 RxJs 订阅答案

【问题标题】：Refactoring chained RxJs subscriptions重构链式 RxJs 订阅
【发布时间】：2021-07-21 15:29:58
【问题描述】：

我有一段代码需要重构，因为它是链式订阅的地狱。

ngOnInit(): void {
    this.dossierService.getIdTree()
        .subscribe(idTree => {
            this.bootstrappingService.refreshObligations(idTree)
                .subscribe(() => {
                    this.dossierPersonsService.retrieveDossierPersons(idTree)
                        .subscribe(debtors => {
                            this.retrieveObligations();
                            this.debtors = debtors;
                        });
                });
        });
}

第一次调用dossierService.getIdTree() 检索idTree，该idTree 被除obligationsService.retrieveObligations() 之外的其他服务使用。
所有服务方法都应该按照它们现在执行的顺序执行。但是retrieveDossierPersons和retrieveObligations可以并行执行。
retrieveObligations() 是订阅另一个 observable 的方法。此方法用于其他几种方法。

我已经对其进行了重构，它似乎可以工作。但是我是否以适当的方式对其进行了重构，或者我的代码可以改进？

   this.dossierService.getIdTree()
      .pipe(
        map(idTree => {
          this.idTree = idTree;
        }),
        switchMap(() => {
          return this.bootstrappingService.refreshObligations(this.idTree)
        }),
        switchMap(
          () => {
            return this.dossierPersonsService.retrieveDossierPersons(this.idTree)
          },
        )
      )
      .subscribe(debtors => {
        this.retrieveObligations();
        this.debtors = debtors;
      });

【问题讨论】：

我不是 map 的这种用法的忠实拥护者 - 它在这里起到了副作用的作用。你的意思是tap 吗？

标签： javascript angular rxjs

【解决方案1】：

类似这样的东西（未检查语法）：

  ngOnInit(): void {
    this.dossierService.getIdTree().pipe(
      switchMap(idTree =>
        this.bootstrappingService.refreshObligations(idTree)).pipe(
          switchMap(() => this.dossierPersonsService.retrieveDossierPersons(idTree).pipe(
            tap(debtors => this.debtors = debtors)
          )),
          switchMap(() => this.retrieveObligations())
        )
    ).subscribe();
  }

使用高阶映射运算符（在本例中为switchMap）将确保订阅和取消订阅内部可观察对象。

在此示例中，您无需单独存储 idTree，因为您可以通过链接管道访问它。

【讨论】：

【解决方案2】：

你可以试试这样的：

 ngOnInit(): void {
   const getIdTree$ = () => this.dossierService.getIdTree();
   const getObligations = idTree => this.bootstrappingService.refreshObligations(idTree);
   const getDossierPersons = idTree => this.dossierPersonsService.retrieveDossierPersons(idTree);

   getIdTree$().pipe(
     switchMap(idTree => forkJoin({
       obligations: getObligations(idTree)
       debtors: getDossierPersons(idTree),
     }))
   ).subscribe(({obligations, debtors}) => {
     // this.retrieveObligations(); // seems like duplicate of refreshObligations?
     this.debtors = debtors;
   });
 }

根据其余代码和模板，您可能还希望通过使用 async 管道来避免解开 debtors

forkJoin 仅在其所有流都完成后才会完成。您可能还想通过管道 catchError 到每个内部 observable 来使用一些错误处理。

您可能想要使用mergeMap 或concatMap（它们采用数组而不是对象）而不是forkJoin——这在很大程度上取决于逻辑和用户界面。 concatMap 将保留序列，mergeMap 不会 - 在这两种情况下，数据可以在到达时累积显示。使用forkJoin，当一个请求卡住时，整个流都会卡住，因此在所有流完成之前，您将无法显示任何内容。

【讨论】：

【解决方案3】：

您可以使用 switchMap 或最好的选择是 concatMap 来确保执行顺序

obs1$.pipe(
 switchMap(data1 => obs2$.pipe(
   switchMap(data2 => obs3$)
 )
)

【讨论】：