Умные буферы

Question

Я отправляю запрос - получаю массив данных. Чтобы управлять этими данными, мне нужно сгладить его, поэтому я могу использовать его как поток объектов, а не поток массива сущностей, но затем по побочному эффекту я хочу, чтобы эти объекты отображались в интерфейсе пользователя одновременно, а не один за другим , поэтому он обновляет только пользовательские интерфейсы.

Скажем, у меня есть такой код:

// this generates a sequence of objects - getCasesStream sends an ajax request, 
// whenever dateRange changes 
casesStm = dateRangeStm.flatMapLatest(getCasesStream) 

casesStm.subscribe((x)=> { console.log(x) }) 

function getCasesStream(dateRange) { 
    return getCases(dateRange.startDate, dateRange.endDate) 
     // api every time returns an array, 
     // but it's difficult to work with array of elements, ergo the flattening 
     .flatMap((x) => x)         
     .filter((x) => _.isNotEmpty(x.value)) 
     .map((caseDoc) => _.assign(caseDoc.value, { 
      key: caseDoc.id 
     })); 
} 

Это работает просто отлично, испускающий одно значение в то время. Теперь я хочу испускать не более 10 значений, если их меньше 10 - испускать все, что осталось.

Я думал, что я мог бы решить, что, делая это:

casesStm 
    .windowWithCount(10) 
    .flatMap((x)=> x.toArray()) 

Но это работает только если getCasesStream (после фильтрации) возвращает по крайней мере 10 пунктов, если она меньше, чем - я не буду даже увидеть их.

Как я могу эффективно использовать элементы здесь? снова:

• апи посылает нам массив
• фильтровать и добавлять дополнительные подпорки к каждому элементу, лучше придавить этот массив
• в конце мне нужно буфера (Дон»(или, может быть, нет?) т хотят заставить браузер перерисовать каждый раз, когда новый элемент приходит)

может быть, я должен использовать общий window, который возвращает длину элементов в getCasesStream, но эта функция не принимает никаких аргументов, как я могу получить длину? Я пробовал использовать windowWithTimeOrCount, который сохраняет испущенный пустой буфер на каждый интервал, даже если нет элементов.

Answer 1

Вы можете отфильтровать эти пустые буферы, и вы также можете просмотреть некоторые из опций, упомянутых в How to create a RxJS buffer that groups elements in NodeJS but that does not rely on forever running interval?.

Представленная идея заключается в использовании оператора buffer с закрывающим селектором. В качестве селектора замыкания вы можете использовать merge(source.skip(9).take(1).repeat(), source.delay(Xms)) (или оператор flatMapFirst, как предлагается в приведенной выше ссылке, просмотрите оба варианта). Таким образом, в принципе, когда нет испускаемого случая, не происходит выброса буфера, и когда приходит случай, оператор merge излучает значение либо в 10-м случае, либо через Xms, в зависимости от того, что наступит раньше. Когда излучается значение от оператора merge, буфер закрывается и испускается.

Вы можете черпать вдохновение из кода здесь:

function emits(who){ 
    return function (x) { console.log([who, "emits"].join(" ") + " " + x + " click(s)");}; 
} 

var Xms = 1700; 

var source = Rx.Observable.fromEvent(document.body, 'click'); 
console.log("running"); 

var delayedSource$ = Rx.Observable.merge(source.skip(9).take(1).repeat(), source.delay(Xms)); 

var buffered$ = source 
    .buffer(function() { return delayedSource$;}).map(function(clickBuffer){return clickBuffer.length;}); 

buffered$.subscribe(emits("buffer")); 

jsbin: http://jsbin.com/siqopuxoli/edit?html,js,console,output

Важное примечание: вы должны также поделиться своим источником случай (если вы не знаете, что это горячий источник уже), а он будет подписан не один раз: getCases(dateRange.startDate, dateRange.endDate).share()