RxJS - как накапливать данные и действовать на нем, без нескольких обратных вызовов

Question

Так что этот код работает, но он стреляет обратный вызов в .flatmap 5 раз:

 var i = 0; 
     const values = {}; 
     return this.obsClient.take(5) 
      .flatMap(v => { 
       const time = Date.now(); 
       values[i] = {time: time, count: v.clientCount}; 
       console.log('values => ', values); 
       i++; 
       return Rx.Observable.timer(100) 
      }); 

«значение => х» получит вход- раз.

Я подумал, что если бы я сделал следующее, это было бы аккумулировать все 5 частей данных перед срабатыванием обратного вызова flatMap:

 var i = 0; 
     const values = {}; 
     return this.obsClient.take(5).takeLast(5) 
     .flatMap(v => { 
      const time = Date.now(); 
      values[i] = {time: time, count: v.clientCount}; 
      console.log('values => ', values); 
      i++; 
      return Rx.Observable.timer(100) 
     }); 

, но он по-прежнему регистрирует «значение => х» в 5 раз. Как я могу накапливать данные и передавать все данные в обратный вызов и не запускать обратный вызов 5 раз?

Answer 1

Если вы хотите только последние пять элементов в последовательности сразу:

return this.obsClient 
       .takeLast(5) 
       .toArray() 
       .flatMap(arr => { /*arr will be an array of the last five items in the sequence */ }) 

Answer 2

Вы могли бы искать bufferCount:

return this.obsClient.bufferCount(5).flatMap(buffer => { 
    console.log(`buffer => ${buffer}`); 
    return Rx.Observable.timer(100); 
}); 

БУФЕРОВ источник Наблюдаемые значения до тех пор, размер не достигнет максимального bufferSize заданного.