CODEWITHSUNDEEP
rxjsrxjs5
James
James

Reputation: 510

Debounce and buffer an rxjs subscription

I have a message queue processor that feeds messages to a service...

q.on("message", (m) => {
  service.create(m)
    .then(() => m.ack())
    .catch(() => n.nack())
})

The service uses an RxJS Observable and subscription to debounceTime() those requests.

class Service {
  constructor() {
    this.subject = new Subject()
    this.subject.debounceTime(1000)
      .subscribe(({ req, resolve, reject }) =>
        someOtherService.doWork(req)
          .then(() => resolve())
          .catch(() => reject())
      )
  }

  create(req) {
    return new Promise((resolve, reject) =>
      this.subject.next({
        req,
        resolve,
        reject
      })
    )
  }
}

The issue is that only the debounced request gets ackd/nackd. How can I ensure that the subscription also resolves/rejects the other requests? bufferTime() gets me a part of the way there, but it does not reset the timeout duration on each call to next().

Upvotes: 5

Views: 3805

Answers (2)

Ignacio Bustos
Ignacio Bustos

Reputation: 1494

For those who are looking for a RXJS 6 Solution, I created a custom operator to behave like a debounce() + buffer() as in the previous answer.

I called it bufferDebounce and the snippet in Typescript with Type inference is here:

import { Observable, OperatorFunction } from 'rxjs'
import { buffer, debounceTime } from 'rxjs/operators'

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();
      },
    }),
  );

You can test its behavior in this example to check if this suits you https://stackblitz.com/edit/rxjs6-buffer-debounce

Upvotes: 11

cartant
cartant

Reputation: 58400

The debounceTime operator that you are currently using can be used to create an observable that can notify buffer of when the current buffer should be closed.

Then, buffer will emit an array of the messages that were received whilst debouncing and you can do with them whatever you want:

this.subject = new Subject();
const closingNotifier = this.subject.debounceTime(1000);
this.subject.buffer(closingNotifier).subscribe(messages => {
  const last = messages.length - 1;
  messages.forEach(({ req, resolve, reject }, index) => {
    if (index === last) {
      /* whatever you are doing, now, with the debounced message */
    } else {
      /* whatever you need to do with the ignored messages */
    }
  });
});

Upvotes: 7

Related Questions