CODEWITHSUNDEEP
rxjsrxjs5
WHITECOLOR
WHITECOLOR

Reputation: 26122

RxJS: (Time) Buffer that starts after next emittion

I wonder how to implement this properly with RxJs (4/5)?

-a-- -b----c----d-----------------------------------------------------------e------f---------------------
-5-sec after-"a"--> [abcd]---new 5 sec timer will start when "e" emited-----5 sec-after-"e"->[ef]-

I think this:

.buffer(source$.throttleTime(5000).debounceTime(5000))

do the job in rxjs 5

Upvotes: 2

Views: 2249

Answers (4)

Ignacio Bustos
Ignacio Bustos

Reputation: 1494

As Trevor mentioned, in RXJS 6 there is no official way but clearly you need to use debounce + buffer in order to achieve that result.

To make things properly, in Typescript and with Type Inference, I created a custom OperatorFunction called bufferDebounce that makes a lot easier to use and understand this operator.

The snippet with type inference

type BufferDebounce = <T>(debounce: number) => OperatorFunction<T, T[]>;

const bufferDebounce: BufferDebounce = debounce => source =>
  new Observable(observer => 
    source.pipe(buffer(source.pipe(debounceTime(debounce)))).subscribe({
      next(x) {
        observer.next(x);
      },
      error(err) {
        observer.error(err);
      },
      complete() {
        observer.complete();
      },
  })
// [as many sources until no emit during 500ms]
source.pipe(bufferDebounce(500)).subscribe(console.log)

You can try it in this working example: https://stackblitz.com/edit/rxjs6-buffer-debounce

Upvotes: 0

Trevor Karjanis
Trevor Karjanis

Reputation: 1724

I am using RxJS 6 and could not readily find the documentation for 5. However, this is a fantastic question. Here was my result which is also demonstrated in a real example reproducing a bug in Angular Material.

source$ = source$.pipe(buffer(source$.pipe(debounceTime(5000))));

Upvotes: 1

corolla
corolla

Reputation: 5656

Having tried all Rxjs 5 buffer variants, in particular bufferTime which emits every n seconds empty or not, I ended up rolling my own bufferTimeLazy:

function bufferTimeLazy(timeout) {
  return Rx.Observable.create(subscriber => {
    let buffer = [], hdl;
    return this.subscribe(res => {
      buffer.push(res);
      if (hdl) return;

      hdl = setTimeout(() => {
        subscriber.next(buffer);
        buffer = [];
        hdl = null;
      }, timeout);

    }, err => subscriber.error(err), () => subscriber.complete());
  });
};

// add operator
Rx.Observable.prototype.bufferTimeLazy = bufferTimeLazy;

// example
const click$ = Rx.Observable.fromEvent(document, 'click');

click$.bufferTimeLazy(5000).subscribe(events => {
  console.log(`received ${events.length} events`);
});

Example: https://jsbin.com/nizidat/6/edit?js,console,output

The idea is to collect events in a buffer and emit the buffer n seconds after first event. Once emitted, empty buffer and remain dormant until next event arrives.

If you prefer not to add operator to Observable.prototype, just invoke the function:

bufferTimeLazy.bind(source$)(5000)

EDIT: Ok, so it's not all bad with Rxjs 5:

var clicks = Rx.Observable.fromEvent(document, 'click').share();
var buffered = clicks.bufferWhen(() => clicks.delay(5000));
buffered.subscribe(x => console.log(`got ${x.length} events`));

Achieves the same. Notice share() to avoid duplicate click subscriptions - YMMV.

Upvotes: 0

Meir
Meir

Reputation: 14375

Your best shot is to use buffer. The buffer has a closing condition, and you'd like a closing condition 5 seconds after a new item was introduced. So, lets suppose you have a source stream, your desired stream will be:

source.buffer(source.throttle(5100).debounce(5000));

This is rxjs 4. I think rxjs has a slightly different buffer operators but the idea is the same.

Explanation: The throttle ensures that for 5100 mSecs you will get only the first "tick". The debounce will propagate this "tick" after 5000 mSecs because there were no other "ticks" since. Note that I chose 5100 mSecs since the timing is not always perfect and if you use 5000 mSecs for both, the debounce might be repeatedly delayed and you'll get starvation. Anyways, your buffer will not loose data, just might group it in chunks bigger than 5000 mSecs.

Rxjs 5 has a bufferToggle operator which might look a better option, yet, the fact that you both open and close the buffer might become risky and make you loose data due to timing issues.

Upvotes: 1

Related Questions