promise - RxJs 缓冲直到数据库插入( promise )

Question

我有一个数据流，数据快速传入。我想通过保持顺序将它们插入数据库。我有一个数据库，它返回一个 promise ，当插入成功时该 promise 将得到解决。

我想创建一个 Rx 流，它缓冲新数据，直到插入缓冲的数据。

我怎样才能做到这一点？

Answer 1

我相信要获得您想要的东西，您需要创建自己的运算符。稍微脱离 RxJS 你可以得到类似的东西(警告，尚未测试)...

export class BusyBuffer<T> {
  private itemQueue = new Subject<T>();
  private bufferTrigger = new Subject<{}>();
  private busy = false;

  constructor(consumerCallback: (items: T[]) => Promise<void>) {
    this.itemQueue.buffer(this.bufferTrigger).subscribe(items => {
      this.busy = true;
      consumerCallback(items).then(() => {
        this.busy = false;
        this.bufferTrigger.next(null);
      });
    });
  }

  submitItem(item: T) {
    this.itemQueue.next(item);
    if(!busy) {
      this.bufferTrigger.next(null);
    }
  }

}

然后可以用作

let busyBuffer = new BusyBuffer<T>(items => {
  return database.insertRecords(items);
});
items.subscribe(item => busyBuffer.submitItem(item));

但这并不完全是纯粹的响应式(Reactive)，有人也许能够想出更好的东西。