使用RxJS向流添加值时触发栈溢出：排查无限循环问题

问题根源 #

原代码触发无限循环的核心原因：

• 每次Apollo推送新会话时，你都会创建一个sessions$的新订阅。
• 由于sessions$是BehaviorSubject的Observable，订阅时会立即发出当前的会话列表值。
• 当你调用setSessions更新列表后，这个订阅会再次收到新的列表值，进而重复执行setSessions([...sessions, newSession])——把同一个新会话反复添加，触发sessions$再次发射值，循环直到栈溢出。

修复方案 #

用RxJS操作符替代嵌套订阅，通过数据流组合的方式管理会话列表，彻底避免循环：

方案1：保留BehaviorSubject（兼容原有依赖） #

private sessionsSubject = new BehaviorSubject<Array<Session> | null>(null);
public sessions$ = this.sessionsSubject.asObservable();

constructor(private apollo: Apollo) {
  this.subscribeToSessions();
}

public subscribeToSessions(): void {
  // 提取Apollo推送的有效新会话流
  const newSession$ = this.apollo.subscribe<SessionCreated>({
    query: sessionOperations.CREATE_SESSION_SUBSCRIPTION
  }).pipe(
    map(({ data }) => data?.sessionCreated),
    filter(Boolean) // 过滤空数据，确保只处理有效会话
  );

  // 组合现有会话流与新会话流，更新列表
  newSession$.pipe(
    withLatestFrom(this.sessions$), // 获取当前最新的会话列表
    map(([newSession, existingSessions]) => {
      return [...(existingSessions ?? []), newSession];
    })
  ).subscribe({
    next: (updatedSessions) => this.setSessions(updatedSessions),
    error: (err) => console.error('会话订阅失败:', err)
  });
}

public setSessions(sessions: Array<Session>): void {
  this.sessionsSubject.next(sessions);
}

方案2：简化为纯Observable流（推荐） #

直接通过RxJS操作符管理状态，去掉手动的BehaviorSubject维护：

// 直接对外暴露组合后的会话列表Observable
public sessions$: Observable<Array<Session>>;

constructor(private apollo: Apollo) {
  this.sessions$ = this.initSessionStream();
}

private initSessionStream(): Observable<Array<Session>> {
  const newSession$ = this.apollo.subscribe<SessionCreated>({
    query: sessionOperations.CREATE_SESSION_SUBSCRIPTION
  }).pipe(
    map(({ data }) => data?.sessionCreated),
    filter(Boolean)
  );

  // 用scan操作符从空数组开始累积新会话
  return newSession$.pipe(
    scan((acc, session) => [...acc, session], [] as Array<Session>),
    startWith([]) // 初始值设为空数组，避免null值
  );
}

关键修改说明 #

• 移除嵌套订阅：不再在Apollo回调里订阅sessions$，避免每次收到新会话就创建新订阅的冗余操作。
• 数据流组合：用withLatestFrom或scan操作符组合现有列表与新会话，确保每次只添加一次新会话。
• 数据过滤：通过filter(Boolean)确保只处理有效的新会话数据，避免无效更新。

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