服务中的 RXJS 订阅链

Question

我有两个服务和一个组件。组件侦听服务 B 的服务 A。我目前正在服务 A 中订阅服务 B，运行过滤功能，然后过滤功能正在触发组件侦听的服务 A 中的 observable。最佳做法是不从服务 A 订阅服务 B，而是将该可观察对象传递给将成为唯一侦听器的组件，但我不知道如何正确处理它。

当前设置：

当用户更改类型时，服务 B 中的订阅会触发服务 A 侦听的订阅，然后获取该数据并创建一个新数组并触发组件侦听的 productsChanged 主题

Service A {
  productsChanged = new BehaviorSubject<Product[]>([]);

  types$: Subscription;

  private json: Product[] = ProductsJSON;
  private currentProducts: Product[] = [];

  constructor(private typeService: TypeService) {
    this.types$ = this.typeService.typeChanged.subscribe((type: Type) => {
      this.filterProducts(type);
    });
  }


  filterProducts(type: Type): void {
    this.currentProducts = [];

    // Do Some Filtering to Set the Current Products Based on Given Type
  }
}

当用户更改产品类型时，它会触发“TypeChanged”，主题 A 会监听以过滤掉与类型不匹配的产品

Service B {
  typeChanged = new BehaviorSubject<Type>(Defaults.TYPE);
  private chosenType: Type;

  constructor() {
    this.typeChanged.pipe(take(1)).subscribe((type) => {
      this.chosenType = type;
    });
  }

  setType(type: Type): void {
    this.chosenType = type;

    this.typeChanged.next(this.chosenType);
  }
}

组件在 HTML 中订阅，只是输出一个产品列表

Component {
  products$: Observable<any>;

  constructor(private productService: ProductService) {
    this.products$ = this.productService.productsChanged.pipe(
      map((product) => {
        return product;
      }),
    );
  }
}

Answer 1

您可以在服务 A 中进行以下更改以实现此目的。

您的 productsChanged 不应再是 BehaviorSubject，而应是 Observable 类型的未分配属性，如下所示：

productsChanged: Observable<Product[]>;

然后在构造函数中执行以下操作：

this.productsChanged = this.typeService.typeChanged.pipe(map((type: Type) => {
  return this.filterProducts(type);
}));

最后你的 filterProducts 方法需要返回过滤后的产品，所以是这样的：

  filterProducts(type: Type): void {
    this.currentProducts = [];

    // Do Some Filtering to Set the Current Products Based on Given Type
    
    return filteredProducts;
  }

Answer 2

我发现有两件事可以改变。

1. 如果您不使用服务 B 中的 chosenType 变量，它可以使您免于在其构造函数中进行额外订阅。由于您已经有一个变量 typeChanged 保存当前类型，因此您可以在需要 chosenType 值的地方使用它。

2. 在服务 A 中，您可以将 productsChanged BehaviorSubject 替换为从服务 B 传递到可观察到的 typeChanged 的函数，并使用 RxJS 运算符 filter 和 switchMap 以及 RxJS 函数 of 到返回一个 Products[] 类型的 observable。

服务A（产品服务）

{
  private json: Product[] = ProductsJSON;
  private currentProducts: Product[] = [];

  constructor(private typeService: TypeService) { }

  productsChanged(): Observable<Products[]> {
    this.currentProducts = [];
    return this.typeService.typeChanged.pipe(
      tap((type: Type) => this.currentProducts = /* something */ ),
      filter((type: Type) => /* filter `type` notifications based on certain conditions */ ),
      switchMap((type: Type) => 
        /* do something and return an observable of type `Products[]` using RxJS `of` function */
      )
    );
  }
}

您可以像现在一样在组件中通过管道传递给它。

组件

{
  products$: Observable<any>;

  constructor(private productService: ProductService) {
    this.products$ = this.productService.productsChanged().pipe(  // <-- function instead of observable
      map((product) => {
        return product;
      }),
    );
  }
}