RXJS: Aggregated debounce

Question

Mój przypadek użycia jest następujący: Otrzymuję zdarzenia, które czasami zdarzają się w seriach. Jeśli pojawi się seria, potrzebuję jej tylko raz. Debounce to robi.

Jednak odrzucenie daje mi tylko ostatni element serii, ale muszę wiedzieć o wszystkich elementach w serii, aby je zebrać (używając mapy płaskiej).

Można to zrobić za pomocą okna czasowego lub bufora, jednak są to ustalone odstępy czasowe, więc w środku serii może wystąpić przekroczenie limitu czasu bufora/okna, w związku z czym dzielenie serii na dwie części w celu obsłużenia zamiast 1.

Więc czego chcę jest coś

. 
. 
event: a 
. 
. -> a 
. 
. 
. 
. 
. 
.event: b 
.event: c 
.event: d 
. 
.-> b,c,d 
. 
. 
. 
. 
.event : e 
. 
. -> e 
. 

Answer 1

ten można osiągnąć z buffer przepuszczając strumień w debounced jako selektor zamykania, np .:

var s = Rx.Observable.of('a') 
    .merge(Rx.Observable.of('b').delay(100)) 
    .merge(Rx.Observable.of('c').delay(150)) 
    .merge(Rx.Observable.of('d').delay(200)) 
    .merge(Rx.Observable.of('e').delay(300)) 
    .share() 
; 

s.buffer(s.debounce(75)).subscribe(x => console.log(x)); 

Oto runnable wersja: https://jsbin.com/wusasi/edit?js,console,output

Answer 2

pewnie szuka bufferWithTimeOrCount

ze strony:

/* Hitting the count buffer first */ 
var source = Rx.Observable.interval(100) 
    .bufferWithTimeOrCount(500, 3) 
    .take(3); 

var subscription = source.subscribe(
    function (x) { 
     console.log('Next: ' + x.toString()); 
    }, 
    function (err) { 
     console.log('Error: ' + err); 
    }, 
    function() { 
     console.log('Completed'); 
    }); 

// => Next: 0,1,2 
// => Next: 3,4,5 
// => Next: 6,7,8 
// => Completed