Flush a RxJs Subject

Question

Я хочу создать тему с трубой bufferTime.например,

subject.pipe(bufferTime(1000, null, this._bufferSize),
  filter((v, i) => {
    return v.length !== 0;
  })
)

После использования этого предмета и завершения работы я бы хотел, чтобы пользователь вызвал метод onComplete / new, который очистит оставшееся содержимое потока.

Так какоснованный на времени, я мог бы подождать, пока поток сам себя очистит, но поскольку я использую AWS Lambda, среда выполнения - это деньги.

Есть ли простой способ реализовать сброс?

Answer 1

Я думаю, что вы ищете takeUntil оператор:

const subject = new Subject();
const complete = new Subject();
const BUFFER_SIZE = 10;

subject
  .pipe(
    takeUntil(complete),
    bufferTime(1000, null, BUFFER_SIZE),
  )
  .subscribe(buffer => {
    console.log(Date.now(), buffer);
  });

Я использую другой Subject, называемый complete, который используется для завершения наблюдаемого и, следовательно, очистки буфера в bufferTime.

См. Рабочий пример здесь: https://stackblitz.com/edit/typescript-ihjbxb