如何做一個滑動緩衝流？

Question

考慮下面的代碼，我使用bufferCount（不太我想要什麼）......

var Rx = require('rxjs/Rx'); 
    var Observable = Rx.Observable; 
    var Subject = Rx.Subject; 

    var first = new Rx.Subject(); 

    var source = first.bufferCount(2).map(a => a.reduce((acc,x) => acc+x,0)); 

    var subscription = source.subscribe(console.log) 

    first.next(1) 
    first.next(2) 
    first.next(2) 
    first.next(3) 

我得到

3 
5 

我希望得到的是

3 
4 
5 

所以緩衝區緩衝了最後2個項目。

有沒有辦法簡單地做到這一點？

Answer 1

您是否嘗試過使用bufferCount的skip/startBufferEvery參數？這允許您生成重疊的緩衝區，然後您可以根據需要進行減少。這種方式不涉及發射值，你必須過濾掉。如果您特別需要緩衝區大小爲2，則可以使用pairwise而不是bufferCount。

基思： - 代碼使用此方法

var Rx = require('rxjs/Rx'); 
var Observable = Rx.Observable; 
var Subject = Rx.Subject; 

var first = new Rx.Subject(); 

var source = first.bufferCount(2,1).map(a => a.reduce((acc,x) => acc+x,0)); 

var subscription = source.subscribe(console.log) 

first.next(1) 
first.next(2) 
first.next(2) 
first.next(3) 

Answer 2

使用scan運算符來保存滑動緩衝區，並對其執行操作。例如：

var source = first.scan((slidingBuffer, newInput) => { 
    return addTo(slidingBuffer, newInput) 
    }, []) 
    .map(processBuffer) 

具有明顯的含義addTo和processBuffer功能。