在 Angular 应用程序中将运算符动态添加到可观察的 Rxjs

Question

在我们的 Angular 应用程序中，我们有一个加载器组件，它通过输入设置器属性接收一个 observable。

在 setter 中，我们首先将布尔值 isLoading 设置为 true，在模板内部开始显示加载微调器。 然后我们订阅 observable，当接收到数据时，isLoading 布尔值再次设置为 false，使微调器消失：

// loading-component template:

<ng-container *ngIf="data; else errorOrLoading">
  ...
</ng-container>

<ng-template #errorOrLoading>
 ...
</ng-template>

// loading-component class (short version without unsubscribe):

  @Input() set data$(data$: Observable<any>) {
    if (data$) {
      this.data = null;
      data$
        .subscribe(data => {
          this.data = data;
        });
    }
  }

如果我们只有一个来自 Observable 的事件，这将非常有效。 但是当 Observable 发出多个事件时，isLoading 不会再次设置为 true，因为没有调用 setter。

有没有办法在给定的 Observable 链开始发出新事件之前动态添加一个额外的点击运算符，允许设置 this.data = null？

例如，如果 Observable 是：

myService.onChanges$.pipe(
  switchMap(() => this.getBackendData())
);

我们可以动态添加一个将管道更改为：

myService.onChanges$.pipe(
  tap(() => this.data = null),
  switchMap(_ => this.getBackendData())
);

更新：我选择简化加载程序控制并将所有可观察的相关逻辑移至服务，这感觉更具可扩展性和灵活性。

Answer 1

更新 #1

解决方案 #1。使用shareReplay()

根据@Reactgular answer

您所要做的就是使用shareReplay() 运算符：

class MyService {
    public data$: Observable<any>;
    public loaded$: Observable<boolean>;

    constructor(private dataService: DataService) {
        this.data$ = this.dataService.loadData().pipe(
            startWith(null), // data$ emit a null before data is ready followed by the API data.
            shareReplay(1);
        );
        this.loaded$ = this.data$.pipe(
           mapTo(true),
           startWith(false)
        );
    }
}

您必须调用 myService.data$.subscribe() 来触发流的第一次读取以使数据准备好。您可以在构造函数中执行此操作，但请记住，Angular 在首次使用之前不会创建服务。如果您希望快速加载数据，请在路由中使用解析器或将服务注入 NgModule 构造函数并在那里订阅。

---

解决方案 #2。使用专属服务

更好的解决方案是引入LoaderService，它将处理组件/视图数据的加载。

根据您的项目需要，它可以是singleton 或shared。

假设我们的服务将只处理当前视图的加载状态（共享服务）

加载器服务：

export class LoaderService {
  private  readonly  _loading: BehaviorSubject<boolean> = new BehaviorSubject<boolean>(false)

  // returns value of `_loading` on the moment of accessing `isLoading`
  get isLoading(): boolean {
    return this._loading.getValue();
  }
  
  // returns value of `_loading` as Observable
  get isLoading$(): Observable<boolean> {
    return this._loading.asObservable();
  }
  
  // pushes `true` as a new value of `_loading` subject and notifies subscribers
  start(): void {
    this._loading.next(true);
  }
  
  // pushes `true` as a new value of `_loading` subject and notifies subscribers
  stop(): void {
    this._loading.next(false);
  }
  
}

假设我们有两个服务：

• API - 仅包含返回纯（未修改）http 流的方法声明
• ComponentService - 在将数据传递给表示组件之前准备数据的服务

我的水果服务

  constructor(
   ..., 
   private api: MyFruitAPIService,
   private loader: LoaderService
  ) { ... }

  getApples(): Observable<T> {
    this.loader.start();
    
    return this.api.getApples()
     .pipe(
      finalize(() => this.loader.stop()) // finalize - will call a function when observable completes or errors
     );
  }
   

MyAppleFruitComponent

 readonly loading$ = this.loader.isLoading$.pipe(tap((loading) => {
   if (loading) {
    this.data = null;
   }
 }));

 constructor(private loader: LoaderService) { ... }

 <ng-container *ngIf="loading$ | async; else errorOrLoading">

 ...

 </ng-container>

Answer 2

简而言之，没有。这是不可能的。这也可能不是一个好主意，因为这样做会破坏封装。

您也不能动态更改函数调用。如果doADance() 为你跳舞，你就不能真正动态地让它添加一个数字列表。函数的实现应该与其调用保持分离。尽管就您的观点而言，Javascript 实际上确实让人们通过绑定不同的上下文等来动态地做一些奇怪的事情。

RxJS 也将实现与流的调用（订阅）分开。如果一个库进行 20 次转换并将该流返回给您，那么您真的不会收到转换列表，这只是库编写者可以在不引入重大更改的情况下更改的实现细节。

更新 1：

的确，封装很重要，而且存在是有原因的。然而，我们 当然可以通过传递动态地将数字列表添加到 doADance() 列表作为参数。也许是一种允许的受控方式 管道内要动态填充的占位符 运营商会是一样的吗？

虽然pipe 中的占位符没有真正意义，因为任何可管道操作符都可以轻松转换为静态操作符，并且任何一组操作符都可以轻松转换为单个操作符。

你能做的是非常接近的事情。例如，这不适用于从库返回的流，但您可以设计流以允许自定义它们的处理方式。

为您服务：

function dataOnChange(): Observable<Data>{
  return myService.onChanges$.pipe(
    switchMap(() => this.getBackendData())
  );
}

function dataOnChangePlus(op): Observable<Data>{
  if(op == null) return dataOnChange();
  return myService.onChanges$.pipe(
    op,
    switchMap(() => this.getBackendData())
  );
}

其他地方：

this.service.dataOnChangePlus(
  tap(_ => this.data = null)
).subscribe(console.log);

在其他地方，做同样的事情，但有点不同：

this.service.dataOnChangePlus(
  st => st.pipe(
    mapTo(null),
    tap(val => this.data = val)
  )
).subscribe(console.log);

现在 dataOnChangePlus 的使用者正在返回一个流，并且还可以帮助定义该流的构造方式。它不是动态添加运算符，但它确实允许您延迟运算符的定义。

好处是每次调用都可以定义不同的东西。

如果您愿意，您可以通过只允许调用者访问特定类型的操作员来缩小调用者的操作范围。例如，只让他们为点击运算符定义 lambda：

function dataOnChangePlusTap(tapper): Observable<Data>{
  return myService.onChanges$.pipe(
    tap(tapper),
    switchMap(() => this.getBackendData())
  );
}

this.service.dataOnChangePlusTap(
  _ => this.data = null
).subscribe(console.log);