angular - 检测主题何时不再有订阅

Question

我正在实现一个 Angular 服务，让消费者根据他们的 id 观察各种值:
它的本质是这样的:

private subjects = new Map<number, Subject<any>>();

public subscribe(id: number, observer: any): Subscription {
  // try getting subject for this id (or undefined if it does not yet exist)
  let subj = this.subjects.get(id);

  // create subject if it does not yet exist
  if (!subj) {
    subj = new Subject<any>();
    this.subjects.set(id, subj);
  }

  // subscribe observer
  const subscription = subj.subscribe(observer);

  // set up teardown logic (gets called when subscription is unsubscribed)
  subscription.add(() => { 
    // remove subject from the map, if this was the last subscription
    if (subj.observers.length == 0) {
      this.subjects.delete(id);
    }
  });

  // return subscription
  return subscription;
}

Here is the full stackblitz example
以上工作正常，但 API 使用起来有点麻烦(在消费者中，我需要手动跟踪所有订阅并确保正确取消订阅)。
我更喜欢有一个返回 Observable 的方法像这样:

public subscribe(id: number): Observable<any> {
  // TODO: Return an observable for this id and make sure that
  // its corresponding subject is in the map iff at least one of the observables
  // for this id has at least one subscription.
  
  return ...;
}

因为这将允许我使用 async 直接从组件模板订阅我需要的值。管道，其中 angular 将负责取消订阅观察者。
但我不太清楚如何实现删除未使用的逻辑 Subject来自 Map当它们不再使用时。有没有好的方法可以做到这一点？
Here is an incomplete stackblitz examples with some test cases

Answer 1

我想你可以尝试这样的事情:

function subscribe(id: number): Observable<any> {

  /* ... */

  return sbj
    .pipe(
      finalize(() => {
        if (subj.observers.length == 0) {
          this.subjects.delete(id);
        }
      })
    );
}

有了这个，您还可以使用 异步 管带 AnonymousSubject返回者 Subject.lift (这是作为 Subject.pipe() 的结果调用的)。 AnonymousSubject确保观察者(例如来自模板)将被添加到``匿名主题 's parent 主题列表。 finalize()在取消订阅源(例如 Subject )时调用。这可能发生在组件被销毁时，或者当 complete 时发生。/ error事件发生，这也包括 Subject 的情况完成。当 Subject完成，它将 send a complete notification to all of its subscribers ，这意味着观察者最终将自动从 Subject 中删除的观察员名单。
编辑
app.component.ts

  show1 = true;
  show12 = true;
  show2 = true;

  v1$: Observable<any>;
  v12$: Observable<any>;
  v2$: Observable<any>;

  constructor(public valueService: ValueService) {
  }

  async ngOnInit() {
    await this.sleep(2000);
    // const s11 = this.valueService.subscribe(1, v => this.v1 = v);
    this.v1$ = this.valueService.subscribe(1);
    await this.sleep(2000);
    // const s21 = this.valueService.subscribe(2, v => this.v2 = v);
    this.v2$ = this.valueService.subscribe(2);
    await this.sleep(2000);
    // const s12 = this.valueService.subscribe(1, () => {});
    this.v12$ = this.valueService.subscribe(1);
    await this.sleep(2000);
    // s12.unsubscribe();
    this.show12 = false
    await this.sleep(2000);
    // s11.unsubscribe();
    this.show1 = false;
    await this.sleep(2000);
    // s21.unsubscribe();
    this.show2 = false
  }

应用程序组件.html

<div *ngIf="show1">
  v1: {{ v1$ | async }}
</div>

<div *ngIf="show12">
  v12: {{ v12$ | async }}
</div>

<div *ngIf="show2">
  v2: {{ v2$ | async }}
</div>

值(value).服务.ts

public subscribe(id: number): Observable<any> {
  let subj = this.subjects.get(id);

  if (!subj) {
    subj = new Subject<any>();
    this.subjects.set(id, subj);
  }

  return subj.pipe(
    finalize(() => {
      if (subj.observers.length === 1) {
        this.subjects.delete(id);
      }
    })
  )
}

StackBlitz .
正如@ggradnig 提到的，支票应该是 subj.observers.length === 1 , 自 finalize() ，至少在 RxJs 6.5.x , 在任何其他取消订阅发生之前运行其回调。