使用响应式扩展定期保存更​​改

Question

我有一个 Angular 应用程序，它有一个非常大的 mat-table。该表的大部分单元格都是mat-select组件（来自Angular Material），当mat-select中的值发生变化时，需要保存，但要遵循以下规则：

1. 每 7 秒保存一次自上次保存以来所做的所有更改。
2. 如果认为某行已完成，则在此时触发保存，包括自上次保存以来所做的所有更改。

我的进攻计划是这样的：

1. 在任何selectionChange 或任何mat-select 上，将整个数据对象发送到Subject。
2. 订阅Subject，通过以下管道运行它：
   • 根据上述条件缓冲可观察流
   • 过滤掉空数组（缓冲区窗口中没有发出任何内容）
   • 将整个更改数组发布到服务器。服务器知道如何处理它。
3. 缓冲区条件本身就是一个 Observable，只要它发出任何东西就会触发缓冲区。
   • 我希望它在 7 秒过后或数据行完成时发出。所以我使用combineLatest 组合两个Observables，一个interval(7000)，以及要保存的数据Subject，过滤后只取完整的数据；并且只取其中一个发出的最后一个。

这是我所做的（简化的）：

my.component.html

  <mat-table>
  <ng-container matColumnDef='someData'>
    <mat-header-cell *matHeaderCellDef>Some Data</mat-header-cell>
    <mat-cell *matCellDef='let row'>
      <mat-form-field>
        <mat-select [(ngModel)]='row.someData' (selectionChange)='dataToSave.next(row)'>
          <mat-option [value]='3'>High</mat-option>
          <mat-option [value]='2'>Medium</mat-option>
          <mat-option [value]='1'>Low</mat-option>
        </mat-select>
      </mat-form-field>
    </mat-cell>
  </ng-container>
  ...
</mat-table>

my.component.ts

import { Component, OnInit, AfterViewInit, OnDestroy } from '@angular/core';
import { DataService } from '../service/data.service';
import { Subject } from 'rxjs/Subject';
import { filter, switchMap, buffer, startWith } from 'rxjs/operators';
import { interval } from 'rxjs/observable/interval';
import { combineLatest } from 'rxjs/observable/combineLatest';

@Component({
  selector: 'app-data-grid',
  templateUrl: './data-grid.component.html',
  styleUrls: ['./data-grid.component.scss']
})
export class DataGridComponent implements OnInit, AfterViewInit, OnDestroy {
  dataToSave = new Subject<any>();

  // trigger the buffer either when 7 seconds has passed, or a row has been completed.
  bufferCondition = combineLatest(
    interval(7000).pipe(
      startWith(0)
    ),
    this.dataToSave.pipe(
      startWith({}),
      filter(row => this.checkAllFieldsCompleted(row))
    )
  );

  constructor(private dataService: DataService) { }

  ngAfterViewInit() {
    this.dataToSave.pipe(
      buffer(this.bufferCondition),
      filter(dataArray => data && dataArray.length > 0),
      switchMap(dataArray => this.dataService.saveData(dataArray))
    ).subscribe(dataSaved => { 
      /* success */ 
    }, err => { 
      /* error */
    });
  }

  checkAllFieldsCompleted(row: any): boolean {
    return /* logic to check if complete */
  }
    ...
}

我认为这将是反应式扩展的完美使用！但我似乎无法让它发挥作用。如果我单独使用 bufferTime 运算符，一切正常。所以我相信错误在于我如何定义自定义缓冲区条件。

Answer 1

经过一番挖掘和反复试验，我有了自己的解决方案。

从this website我了解到

请注意，combineLatest 在每个 observable 发出至少一个值之前不会发出初始值。

在我的缓冲条件下，我试图结合两个可观察的：7 秒的间隔，以及当一行完成时。我注意到预期的行为仅在我第一次完成一行之后才开始 - 只有这样它才会开始每 7 秒保存一次。问题是dataToSave observable 上的filter 运算符。 combineLatest 直到 两个 observables 都发出了一些东西，才发出任何东西。

我更改了这段代码：

bufferCondition = combineLatest(
  interval(7000).pipe(
    startWith(0)
  ),
  this.dataToSave.pipe(
    startWith({}),
    filter(row => this.checkAllFieldsCompleted(row))
  )
);

到这里：

bufferCondition = combineLatest(
  interval(7000),
  this.dataToSave.pipe(
    filter((row, index) => index === 0 || this.checkAllFieldsCompleted(row))
  )
);

而且效果很好！基本上，我不需要 startWith 运算符，在 filter 上，我让 observable 发出对数据的第一个更改，但在一行完成之前没有其他更改。