RXJS: Aggregated debounce

Question

My use case is as following: I get events, which sometimes happen in bursts. If a burst occurs, I only need to handle it once though. Debounce does this.

However, debounce only gives me the last element of a burst, but I need to know about all elements in a burst to aggregate on them (using flatmap).

This could be done by a timed window or buffer, however, these are fixed intervals, so a buffer/window timeout could occur in the middle of a burst, therefore splitting the burst in 2 parts to handle instead of 1.

So what I'd want is something like

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

did you manage to solve your issue? – user3743222 Mar 12 '16 at 19:20 — user3743222, Mar 12 '16 at 19:20
@user3743222 The answer of Matt burnell works for me. – Seba Kerckhof Mar 14 '16 at 13:38 — Seba Kerckhof, Mar 14 '16 at 13:38

Matt Burnell · Accepted Answer · 2016-03-13T10:56:05.343

24

This can be achieved with buffer by passing a debounced stream in as a closing selector, e.g.:

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));

Here's a runnable version: https://jsbin.com/wusasi/edit?js,console,output

edited Mar 13 '16 at 10:56

answered Mar 02 '16 at 06:06

Matt Burnell

2,646
1
22
23

1

I had to make some changes to get this to work in RxJS 5: http://jsbin.com/pubowum/2/edit?html,js,console,output – Oliver Joseph Ash May 25 '17 at 07:33
For rxJava at least, this re-emits the same list every 75 ms – Code Wiget Sep 21 '20 at 22:24
its perfect solution – shutsman May 04 '22 at 13:07

score -1 · Answer 2 · answered Mar 01 '16 at 13:27

-1

You're probably looking for bufferWithTimeOrCount

from the page:

/* 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

answered Mar 01 '16 at 13:27

kakigoori

1,218
10
20

1

No, buffer with count does something different, in my initial post I explained why buffer doesn't work, and it's the same if you buffer with time, count or both. In short: buffer (and each variant) can emit observables in the middle of a burst. This is different from debounce, which will only emit after an amount of time without incoming events. – Seba Kerckhof Mar 01 '16 at 15:07

RXJS: Aggregated debounce

2 Answers2

Linked