Angular 11中使用RxJS设置轮询定时器避免流类型错误的方法

• 直接将类方法onBlur传入window.addEventListener绑定为全局事件回调，未处理this指向问题：事件触发时回调内的this指向全局window对象，代码中读取的this.stopPolling值为undefined。
• RxJS 6.6.7版本中takeUntil操作符要求传入合法的Observable类流对象，接收到undefined参数时就会抛出你看到的"提供了undefined作为流"的类型错误，和报错栈的触发位置完全匹配。
• 原有逻辑未做防重复处理，每次触发blur事件都会重新创建轮询流，存在重复发起请求、内存泄漏的隐患，也未在组件销毁阶段清理副作用。

前置类属性定义 #

提前初始化所有流依赖的对象，避免运行时出现undefined传参：

import { Subject, Observable, timer } from 'rxjs';
import { switchMap, retry, shareReplay, takeUntil } from 'rxjs/operators';

// 组件类内声明
private stopPolling$ = new Subject<void>();
private trendingSet$: Observable<TrendingSet>;

// 类属性形式定义事件回调，自带this绑定，可直接用于全局事件监听/移除
private onWindowBlur = (): void => {
  // 避免重复创建轮询流
  if (this.trendingSet$) return;
  this.trendingSet$ = timer(1, HotComponent.REFRESH_INTERVAL_IN_MS).pipe(
    switchMap(() => this.apiService.getTrending()),
    // 配置重试规则，接口偶发失败时不直接终止轮询
    retry({ count: 2, delay: 1000 }),
    // 多订阅场景下共享上游请求结果，减少重复调用
    shareReplay({ bufferSize: 1, refCount: true }),
    // 传入合法的Observable作为流终止信号
    takeUntil(this.stopPolling$)
  );
  this.trendingSet$.subscribe((result: TrendingSet) => {
    this.articleStats = result.trending_articles;
  });
}

生命周期内的副作用管理 #

• 初始化阶段绑定全局blur事件：

ngOnInit(): void {
  window.addEventListener('blur', this.onWindowBlur, false);
}

• 组件销毁阶段统一清理所有副作用，避免后台残留请求：

ngOnDestroy(): void {
  // 触发终止信号，停掉所有轮询流
  this.stopPolling$.next();
  this.stopPolling$.complete();
  // 移除全局事件监听
  window.removeEventListener('blur', this.onWindowBlur, false);
}

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