从 Observable 更新数据时应用角度过滤器

Question

我有一个 Angular 13 应用程序，它是一种监控不断变化的数据的仪表板。数据来自 Websocket，并实时添加/更新表中的记录。该表有一些静态工作的基本过滤器选项（我也想合并一个排序选项），但是应用过滤器时，传入的新数据不会反映在过滤结果中。

我的目标是即使应用过滤器（以及后来的排序），也能适当更新表。

相关代码如下：

component.ts

myDataStream: Subject<DataStreamRecord[]> = new Subject();
records: Array<DataStreamRecord> = [];

ngOnInit(): void {

    // init stream listener
    this.myDataStream.subscribe({
      next: (data) => { 
        data.forEach(item => {
          // new item?
          var idx = this.records.findIndex(e => { return e.id === item.id });
          const now: any = new Date();
          const newItem = Object.assign({ ts: Math.trunc(now.getTime()/1000) }, item);
          if (idx >= 0) { // not new, update
            this.records[idx] = newItem;
          } else { // new item, add
            this.records.push(newItem);
            idx = this.records.length-1;
          }
        })
      },
      error: (err) => { console.error(err); },
      complete: () => { 'myDataStream complete' }
    });

}

component.html

<table class="table table-striped">
    <thead>
        <tr>...</tr>
    </thead>
    <tbody>
        <tr *ngFor="let record of records | recordFilter:form.value">...</tr>
        ...

recordfilter.pipe.ts

export class RecordFilterPipe implements PipeTransform {

  transform(list: DataStreamRecord[], filters: Object) {
    let selectedAttribute1: Attribute1 = _.get(filters, 'selectedAttribute1');
    let selectedAttribute2: Attribute2 = _.get(filters, 'selectedAttribute2');
    if (!selectedAttribute1 && !selectedAttribute2) return list;

    return list.filter(item => {
      // provider
      if (!selectedAttribute1) return item;
      return item.value1 === selectedAttribute1.value
    }).filter(item => {
      // race
      if (!selectedAttribute2) return item;
      return item.value2 === selectedAttribute2.value;
    })
  }

}

我知道 PipeTransform 返回主数组的快照，这当然就是为什么添加到数组的新记录没有反映在转变。我预计我的方法需要编辑，因为我已经审查了其他 SO 问题，例如 这个，我感觉我的 Observable 不太正确（无论如何支持我的用例），正如这篇（和其他）帖子一样指的是可观察数组，这不是我这里所拥有的......除非我有？