How can I detect when an rxjs subscriber unsubscribes?

Question

I need to run some clean-up code whenever the last subscriber of a subject unsubscribes. How would I do that?

This is what I have so far, but it seems to me to not be a good way of doing it. Surely there is a reactive way of doing this?

export class TestSubject<T> extends BehaviorSubject<T> {
  protected _subscribe(subscriber: Subscriber<T>): Subscription {
    const originalSubscription = super._subscribe(subscriber);
    return new TestSubjectSubscription<T>(
      this,
      subscriber,
      originalSubscription,
      (observer) => this.registerUnsubscription(observer));
  }

  private registerUnsubscription(subscriber: Observer<T>) {
  }

}

export class TestSubjectSubscription<T> extends SubjectSubscription<T> {
  constructor(
    subject: TestSubject<T>,
    subscriber: Observer<T>,
    private inner: Subscription,
    private notifyUnsubscribed: (subscriber: Observer<T>) => void) {
    super(subject, subscriber);
  }

  unsubscribe() {
    this.inner.unsubscribe();
    this.notifyUnsubscribed(this.subscriber);
  }
}

Answer 1

.finally() can be used to invoke a callback after the stream completes (error/complete/unsubscribe)