RXJS中如何修改Observable返回的每个结果值？代码求助

我有一个reloadProducts函数，用于重新加载购物车中的商品。当前存在CartProduct和Product两个模型，可通过CartProduct的product_name调用服务方法获取对应的Product。

需要在映射CartProduct时，根据CartProduct的mixMatch布尔值，为对应的Product设置mixMatch属性：若CartProduct的mixMatch为true，则将Product的mixMatch设为true；反之设为false。但在代码标注#HERE处的实现未能成功，求指导。

原代码 #

@HostListener('window:storage')
reloadProducts() {
  this.cartProducts = Array.from(this.cartService.getProducts());
  this.products = undefined;
  this.totalPrice = undefined;
  this.producers = undefined;
  this.producerNames = [];

  forkJoin(
    this.cartProducts.map(cartProduct => {
      let result = this.productService.getProduct(cartProduct.product_name);
       // HERE
      result.subscribe((product) => {
        if (cartProduct.mixMatch) {
          product.mixMatch = true;
        }
        else if (!cartProduct.mixMatch) {            
          product.mixMatch = false;
        }
      })
      return result;
    })
  ).pipe(
    switchMap(productListDto => {
      productListDto.map(product => {
        if (this.producerNames.indexOf(product.producer_name) === -1) {
          this.producerNames!.push(product.producer_name)
        }
      })
      this.products = productListDto
      this.mixMatchProducts = productListDto.filter(p => p.mixMatch)

      return forkJoin(
        this.producerNames.map(producerName => this.producerService.getProducerByName(producerName))
      )
    })
  ).subscribe(producers => this.producers = producers)
  this.fetchTotalPrice();
}

问题原因 #

你直接订阅了result这个Observable，这会导致forkJoin最终拿到的productListDto还是服务返回的原始Product对象——内部订阅里修改的是另一个流中的对象引用，不会同步到forkJoin收集的结果数组中。同时这种写法还会造成不必要的订阅泄漏。

解决方案 #

改用map操作符在Observable数据流中直接修改Product对象，这样forkJoin收集到的就是已经设置好mixMatch属性的结果：

修改后的forkJoin内部映射逻辑如下：

forkJoin(
  this.cartProducts.map(cartProduct => {
    // 用pipe+map在数据流中修改属性
    return this.productService.getProduct(cartProduct.product_name).pipe(
      map(product => {
        // 简化赋值逻辑，直接取cartProduct的mixMatch值
        product.mixMatch = cartProduct.mixMatch;
        return product;
      })
    );
  })
)

修改说明 #

1. 通过map操作符在Observable的数据流中修改Product对象，确保forkJoin返回的productListDto是已经更新后的对象数组。
2. 简化了赋值逻辑，直接将cartProduct.mixMatch的值赋给product.mixMatch，无需多余的条件判断。
3. 移除了不必要的内部订阅，避免了订阅泄漏问题，所有Observable的订阅由forkJoin统一管理。

修改后，后续switchMap中的productListDto就会包含正确的mixMatch值，this.mixMatchProducts的过滤逻辑也能正常工作。

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