RxJS - какой оператор может объединить историю последних N событий?

Question

Я ищу оператора, который преобразует поток в поток с историей или последними событиями N.

Итак, для потока, например, такого как:

clicks$.subscribe((ev) => ...)

Я быхотелось бы иметь что-то вроде:

clicks$.pipe(
   historyOf(2)
).subscribe(([ev, ev-1, ev-2]) => ...)

Существует ли такой оператор или существует простой способ объединить существующие операторы для достижения желаемого эффекта?

Answer 1

Я запустил (очень) небольшую библиотеку некоторых утилит rxjs, которые я хотел.Бывает, что есть тот, который может вас заинтересовать: withHistory.Из документов:

source:         -1----2------3--------4--------5--------|
withHistory(2): -[1]--[2,1]--[3,2,1]--[4,3,2]--[5,4,3]--|
withHistory(0): -[1]--[2]----[3]------[4]------[5]------|

Библиотека находится здесь: https://github.com/simontonsoftware/s-rxjs-utils

Answer 2

В идеале вы можете использовать bufferCount или, в зависимости от того, как вы хотите обработать начало или окончание последовательности, когда также меньше N событий scan или filter:

const N = 3;

of(1, 2, 3, 4, 5, 6, 7).pipe(
  bufferCount(N, 1), // `1` means it'll emit after every emission from source
).subscribe(console.log);

... или если вы хотите пропустить выбросы после завершения цепочки:

of(1, 2, 3, 4, 5, 6, 7).pipe(
  bufferCount(N, 1),
  filter(arr => arr.length === N),
).subscribe(console.log);

... или если вы хотите также начать события:

of(1, 2, 3, 4, 5, 6, 7).pipe(
  scan((acc, item) => [...acc, item].slice(-N), []),
).subscribe(console.log);

Демонстрация в реальном времени: https://stackblitz.com/edit/rxjs-xtnscq