Using rxjs to throttle a callback

Question

I am consuming an API in which I register a callback that occurs frequently.

function myCallback(event) {
  // do things with event, something computationally intensive
}

const something = createSomething({
  onSomethingHappened: myCallback
})

I'd like to limit the rate at which this callback fires, probably using throttle. This project uses Angular which bundles rx. How can I adapt my code so myCallback it throttled at 300ms using rx?

I have a basic grasp on how observables work but it's been a bit confusing to figure out how the callback interface would convert to observable interface.

(edited as answers come)

Answer 1

I think you can use fromEventPattern:

let something;

const src$ = fromEventPattern(
  handler => (something = createSomething({ onSomethingHappened: handler })),
);

src$.pipe(
  throttleTime(300),
  map(args => myCallback(args))
);

Note: this assumed that myCallback is a synchronous operation.

The first argument passed to fromEventPattern is the addHandler. It can also have the removeHandler, where you can put your teardown logic(e.g: releasing from memory, nulling out values etc).

---

In order to get a better understanding of what is handler and why is it used there, let's see how fromEventPattern is implemented:

return new Observable<T | T[]>(subscriber => {
  const handler = (...e: T[]) => subscriber.next(e.length === 1 ? e[0] : e);

  let retValue: any;
  try {
    retValue = addHandler(handler);
  } catch (err) {
    subscriber.error(err);
    return undefined;
  }

  if (!isFunction(removeHandler)) {
    return undefined;
  }

  // This returned function will be called when the observable
  // is unsubscribed. That is, on manual unsubscription, on complete, or on error.
  return () => removeHandler(handler, retValue) ;
});

Source.

As you can see, through handler, you can let the returned observable when it's time to emit something.

Answer 2

You can just pipe the operator throttleTime to a fromEvent stream.

import { fromEvent } from 'rxjs';
import { throttleTime } from 'rxjs/operators';

const mouseMove$ = fromEvent(document, 'mousemove');
mouseMove$.pipe(throttleTime(300)).subscribe(...callback);