Взять, пока N одинаковых номеров с RXJS?

Question

У меня есть код, который генерирует значения с определенной скоростью:

let source = zip (
             interval(150) ,
              from ([0,1,1,1,1,2,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3]) 
            )  .subscribe(a=>console.log(a))

Я хочу найти значение - после того, как N значения были установлены (что означает - N повторяющиеся идентичные значения).

Например, если N=5, тогда мне нужно найти первый раз, когда он выдает 5-идентичные значения. (и выбросить это значение.)

Вопрос:

Как я могу обнаружить N идентичных значений и выдать это значение ( и завершить поток )?

Итак, ожидаемое значение здесь - 3, поскольку это первая серия из 5 одинаковых значений:

Демо онлайн

Answer 1

это немного некрасиво, но это будет работать

from([0,1,1,1,1,2,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3]).pipe(
   scan((acc, cur) => {
     if (cur !== acc.lastValue) {
       acc.count = 0;
       acc.lastValue = cur;
     } else {
       acc.count++;
       acc.lastValue = cur;
       if (acc.count >= acc.nValues) {
         acc.found = true;
       }
     }
     return acc;
   },
   {
      count: 0,
      lastValue: undefined,
      nValues: 5,
      found: false
   }),
   filter(result => result.found),
   map(result => result.lastValue),
   take(1)
).subscribe(console.log)

Answer 2

Вы можете использовать bufferCount(N, 1), где 1 указывает bufferCount выдавать свой буфер после каждого излучения, что означает, что он будет выдавать массивы N элементов.Затем просто убедитесь, что все они одинаковые и полные.

const N = 5

from ([0,1,1,1,1,2,3,3,3,3,3,3,3,3,3,3,3,3,3,3,3]) 
  .pipe(
    bufferCount(N, 1),
    filter(buffer => buffer.length === N && buffer.every(i => i === buffer[0])),
    take(1),
    map(buffer => buffer[0]),
  )
  .subscribe(console.log);

Демонстрационная версия: https://stackblitz.com/edit/rxjs6-demo-xzb5so?file=index.ts