自定义rxjs运算符(缓冲区)阻止Angular应用程序保持稳定

我正在开发一个自定义rxjs操作符来扩展 bufferTime 操作员功能。 BufferTime需要时间跨度和缓冲区的最大大小(可选),然而,在我的情况下,我需要重置缓冲区,而不仅仅是基于时间&大小,但也对额外的触发-routeChange。

我已经创建了一个自定义操作符来处理这个问题,它需要timeSpan、maxSize和trigger$,为timeSpan buffer和size buffer创建可观察的对象,最后将这三个对象合并在一起,并作为 buffer 操作人员

以下是源代码:

function bufferFrom<T>(...triggers: Observable<any>[]): (source$: Observable<T>) => Observable<T[]> {
   // restart Subject used for taking control over all triggers and resetting them on emit
   const restart: Subject<void> = new ReplaySubject<void>(1);

   // function that wraps trigger with restart Observable to take control over that
   const withRestart: <R>(source: Observable<R>) => Observable<R> = <R>(source: Observable<R>) => restart.asObservable().pipe(switchMap(() => source));

   // create Observable from all triggers connected with restart control
   const bufferTrigger = merge(...triggers.map(withRestart));

   // emit first reset to enable Observables initially
   restart.next();
 
   // return the source$ Observable with buffer, and reset applied after buffer is emitted
   return source$ => source$.pipe(
      buffer(bufferTrigger),
      tap(() => restart.next()),
      filter(items => items.length > 0),
   );
}

export function bufferAll<T>(bufferTimeSpan: number, bufferSize: number, trigger$: Observable<any>): (source$: Observable<T>) => Observable<T[]> {
   return (source$) => {
      const size$ = source$.pipe(bufferCount(bufferSize));
      const timeSpan$ = source$.pipe(bufferTime(bufferTimeSpan * 1000));

      return source$.pipe(bufferFrom(size$, timeSpan$, trigger$));
   };
}

我的问题是,在Angular应用程序中,在根目录中提供的服务中使用这样的操作符会导致应用程序不稳定,从而阻碍应用程序的渲染。

下面是我如何尝试在角度区域之外运行代码,但在我的情况下,这不起作用。然而,如果我使用一个简单的 缓冲时间 与我的实现不同,它与下面的代码配合得很好。

 this.zone.runOutsideAngular(() => {
    const routeChanges$ = this.router.events.pipe(filter(event => event instanceof NavigationEnd));

    this.dataSubject.asObservable().pipe(
      // Send after 60 seconds or when 10 items in buffer or when route changes
      bufferAll(60, 10, routeChanges$),
    ).subscribe((data) => {
      this.zone.run(() => console.log(data));
    });
});

我不知道为什么我的操作符会阻止Angular 缓冲时间 但事实并非如此,因为它们都在基于时间/间隔创建无限的可观测值,而根据角度文档 https://angular.io/api/core/ApplicationRef#isstable-examples-and-caveats 这可能是一个阻碍渲染和应用程序稳定状态的问题。

你知道我的实现中可能有什么错误吗?我错过了什么?

补充了一个简短的问题 stackblitz reproduction

数据服务ts包括自定义操作员代码以及 缓冲时间 或 bufferCount 接线员。使用时 缓冲时间 或 缓冲区计数 operators应用程序是稳定的,如果被注释掉并切换到my operator,它将变得永久不稳定。