RxJS timeout是否自动取消订阅？NestJS缓存Observable疑问

在NestJS中，我通过对Observable使用cache来减少Firestore的调用次数，同时利用Observable监听Firestore的实时变更。由于页面会调用多个API接口，我希望减少对Firestore同一文档的读取次数，并且在该文档10分钟无调用后销毁对应的Observable。

我已测试使用cache-manager存储Observable，功能正常，但疑惑当timeout触发时，是否会自动关闭订阅（取消订阅）。测试发现似乎并未取消订阅，这会导致内存泄漏吗？有没有办法在timeout触发时取消订阅？我尝试过触发Subject并使用takeUntil操作符，请问如何确认订阅已取消？

let observable: Observable<DocumentSnapshot<T>> = await this.cacheManager.get(key);
    
if(!observable) {
  const subject = new Subject<void>();

  observable = new Observable<DocumentSnapshot<T>>(obs => {
    doc.onSnapshot(snapshot => obs.next(snapshot));
  }).pipe(
    timeout({
      each: 5000,
      with: () => {
        this.cacheManager.del(key);
        subject.next();
        throw new Error('Timeout');
      }
    }),
    shareReplay(1),
    takeUntil(subject),
  );

  await this.cacheManager.set(key, observable);
}

return observable;

1. timeout触发时是否自动取消订阅？ #

你当前的写法里，timeout抛出错误后会终止Observable流，但不会自动取消Firestore的onSnapshot订阅——因为自定义Observable里没有返回清理逻辑。Firestore的onSnapshot会一直保持监听，直到你调用它返回的unsubscribe函数，这种情况确实会导致内存泄漏。

2. 如何在timeout时取消订阅？ #

修改自定义Observable的创建逻辑，保存onSnapshot返回的unsubscribe函数，并在Observable的清理回调中执行它。同时结合takeUntil和finalize操作符，确保超时或主动取消订阅时能彻底清理Firestore监听：

let observable: Observable<DocumentSnapshot<T>> = await this.cacheManager.get(key);
    
if(!observable) {
  const subject = new Subject<void>();

  observable = new Observable<DocumentSnapshot<T>>(obs => {
    // 保存Firestore的unsubscribe函数
    const unsubscribe = doc.onSnapshot(snapshot => obs.next(snapshot));
    
    // Observable被取消订阅时执行清理
    return () => {
      unsubscribe();
      subject.complete(); // 完成Subject避免内存泄漏
    };
  }).pipe(
    timeout({
      each: 600000, // 改为需求的10分钟
      with: () => {
        this.cacheManager.del(key);
        subject.next();
        throw new Error('Document watch timed out due to inactivity');
      }
    }),
    shareReplay(1),
    takeUntil(subject),
    finalize(() => {
      // 最终清理：确保Subject被完成
      if (!subject.closed) {
        subject.complete();
      }
    })
  );

  await this.cacheManager.set(key, observable);
}

return observable;

3. 如何确认订阅已取消？ #

有几种直观的验证方式：

• 添加日志：在Firestore的unsubscribe函数和Observable的清理回调中加入日志，比如console.log('Firestore watch unsubscribed')，超时或取消订阅时会输出对应日志。
• Firebase控制台查看：进入Firebase控制台的Firestore页面，在「监控」->「实时监听」中可以看到当前活跃的监听连接，确认目标监听是否已被移除。
• 内存快照检测：使用Chrome DevTools的「内存」面板，分别记录订阅前、订阅后、超时后的堆快照，对比查看是否有未释放的Firestore相关对象。

另外注意：你当前代码中timeout的each参数设置为5000（5秒），和需求的10分钟不符，记得改为600000毫秒。

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