RxJS - взять последние элементы из наблюдаемого

Я хочу взять 3 последних элемента из наблюдаемого. Скажем, что моя временная шкала выглядит так:

--a---b-c---d---e---f-g-h-i------j->

где: a, b, c, d, e, f, g, h, i, j are emitted values

Всякий раз, когда выдается новое значение, я хочу получить его немедленно, чтобы он выглядел так:

[a]
[a, b]
[a, b, c]
[b, c, d]
[c, d, e]
[d, e, f]
[e, f, g]
[f, g, h]
... and so on

Я думаю, что это супер полезно. Представьте, что вы создаете чат, где вы хотите отобразить 10 последних сообщений. Когда приходит новое сообщение, вы хотите обновить свое представление.

Моя попытка: демонстрация

Ответ 1

Вы можете использовать scan для этого:

from(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u'])
  .pipe(
    scan((acc, val) => {
      acc.push(val);
      return acc.slice(-3);
    }, []),
  )
  .subscribe(console.log);

Это напечатает:

[ 'a' ]
[ 'a', 'b' ]
[ 'a', 'b', 'c' ]
[ 'b', 'c', 'd' ]
[ 'c', 'd', 'e' ]
...
[ 's', 't', 'u' ]

bufferCount не будет делать то, что вы хотите. Он будет излучать только тогда, когда каждый буфер равен точно === 3 что означает, что вы не получите никакого излучения, пока не отправите хотя бы 3 сообщения.

Ответ 2

Вы можете посмотреть Наблюдаемый # bufferCount. Одно отличие состоит в том, что он хочет как минимум 3 раза испускать (первый параметр в этом примере).

const source = Rx.Observable.interval(1000);
const example = source.bufferCount(3,1)
const subscribe = example.subscribe(val => console.log(val));

<script src="https://unpkg.com/@reactivex/[email protected]/dist/global/Rx.js"></script>