如何链接 Angular 可观察量？

Question

我的组件需要在发出 API 请求之前检查是否设置了某些应用程序首选项。现在我已经这样设置了，我以 2 分钟的计时器更新组件的数据：

ngOnInit(): void {
    this.subscription = timer(0, 120 * 1000).subscribe(() => {
        this.shopService.getPreferencesAsObservable().subscribe(preferences => {
            if(preferences) {
                this.getInitialPendingSlotsOrders();
                this.getInitialPendingNoSlotsOrders();
            }
        });
    });
}

getInitialPendingSlotsOrders(){
    this.apiService.fetchShopOrders("status=PENDING&only_slots=true").subscribe(orders=> {
        /* do stuff with orders */
        /* it can happen that I need to get another page of data */
        if(orders.next_page){
            this.getNextSlotsSetOfPendingOrders();
        }
    });
}

getInitialPendingNoSlotsOrders(){
    this.apiService.fetchShopOrders("status=PENDING").subscribe(orders=> {
        /* do stuff with orders */
        /* it can happen that I need to get another page of data */
        if(orders.next_page){
            this.getNextNoSlotsSetOfPendingOrders();
        }
    });
}

getNextSlotsSetOfPendingOrders() { 
    this.apiService.fetchNextSetOfOrders(this.nextSlotsUrl).subscribe(nextSetOfOrders => {
        /* do stuff with next set of orders */
    })
}

getNextNoSlotsSetOfPendingOrders() { 
    this.apiService.fetchNextSetOfOrders(this.nextNoSlotsUrl).subscribe(nextSetOfOrders => {
        /* do stuff with next set of orders */
    })
}

我认为这会起作用，但我已经遇到了这样的情况：我看到正在进行一些额外的 API 调用。我知道这与链接可观察量有关。我可以做什么来清理这个？

先感谢您。

Answer 1

您有多个嵌套订阅。它们会导致多个可能永远不会关闭的开放订阅。相反，您需要使用各种可用的 RxJS 运算符将其限制为单个订阅。

看到你需要并行触发两个请求，你也可以使用 RxJSforkJoin函数。

请参阅此处快速了解。

简而言之

• switchMap运算符将一个可观察值映射到另一个可观察值
• filter运算符根据条件继续运算符链
• forkJoin并行组合并触发多个可观察量

尝试以下操作

ngOnInit(): void {
  this.subscription = timer(0, 120 * 1000).pipe(
    switchMap(() => this.shopService.getPreferencesAsObservable()),
    filter(preferences => !!preferences) // emit only if `preferences` is defined and truthy
    switchMap(() => 
      forkJoin({
        slots: getInitialPendingOrders(true),
        noSlots: getInitialPendingOrders(false)
      })
    )
  ).subscribe({
    next: ({ slots, noSlots }) => {
      // do stuff with orders from `slots` and `noSlots` responses
    },
    error: (error: any) => {
      // handle error
    }
  });
}

getInitialPendingOrders(slots: boolean): Observable<any> {
  return this.apiService.fetchShopOrders("status=PENDING" + slots ? "&only_slots=true" : '');
}

更新

根据经验，您应该返回可观察值，并仅在需要响应时进行订阅。在您的情况下，您可以将 a 传递switchMap给 的每个参数forkJoin，并有条件地返回一个可观察的值。当您不希望返回任何内容时，返回 RxJS 常量EMPTY会从forkJoin. 请注意，forkJoin只有当所有源可观察值完成时才会发出。

ngOnInit(): void {
  this.subscription = timer(0, 120 * 1000).pipe(
    switchMap(() => this.shopService.getPreferencesAsObservable()),
    filter(preferences => !!preferences) // emit only if `preferences` is defined and truthy
    switchMap(() => 
      forkJoin({
        slots: getInitialPendingOrders(true).pipe(
          switchMap((orders: any) => {
            /* do stuff with orders */
            return orders.next_page ? this.getNextSlotsSetOfPendingOrders() : EMPTY;
          })
        ),
        noSlots: getInitialPendingOrders(false).pipe(
          switchMap((orders: any) => {
            /* do stuff with orders */
            return orders.next_page ? this.getNextNoSlotsSetOfPendingOrders() : EMPTY;
          })
        )
      })
    )
  ).subscribe({
    next: ({ slots, noSlots }) => {
      // do stuff with next set of orders from `slots` and `noSlots`
    },
    error: (error: any) => {
      // handle error
    }
  });
}

getInitialPendingOrders(slots: boolean): Observable<any> {
  return this.apiService.fetchShopOrders("status=PENDING" + !!slots ? "&only_slots=true" : '');
}

getNextSlotsSetOfPendingOrders(): Observable<any> { 
  return this.apiService.fetchNextSetOfOrders(this.nextSlotsUrl);
}

getNextNoSlotsSetOfPendingOrders(): Observable<any> { 
  return this.apiService.fetchNextSetOfOrders(this.nextNoSlotsUrl);
}