rxjs-audit-with

An RxJS operator that serializes async calls in Observables by keeping the latest values and dropping interim ones while the async call is blocked. It's called auditWith because it's like RxJS's audit, which only keeps the latest value in the pipe, but it feeds that value through the pipe only when the async callback is free again.

import auditWith from "rxjs-audit-with";

async function slowFunc(num: number) {
  await new Promise(resolve =>
    setTimeout(() => {
      console.log("Processed num", num);
      resolve(true);
    }, 3000),   // deliberately slow, taking 3 secs
  );
}

interval(500)
  .pipe(
    take(13),
    auditWith(slowFunc),  // Calls slowFunc with latest value once it's free
  )
  .subscribe();

// Sample output:
// Processed num 0
// Processed num 5
// Processed num 11
// Processed num 12

The most common use cases for this are where you have an async function (e.g. web fetch) where you'd like to serialize calls to it (i.e. it shouldn't be concurrent), but you only want to call it with the most recent value generated by the Observable. It can also act as a semaphone to protect data structures that can't be re-entrant across threads.

Install

RxJS is required as a peer dependency, but you no doubt already have it installed if you're looking at this package. But if not, you should use these instructions to install it.

Then:

npm install rxjs-audit-with

or

yarn add rxjs-audit-with

How It Works

unction auditWith<T>(callback: (value: T) => Promise<any>):
  MonoTypeOperatorFunction<T> {
  const freeToRun = new BehaviorSubject(true);

  return (source: Observable<T>) => {
    return source.pipe(
      audit(val => freeToRun.pipe(filter(free => free))),
      tap(() => freeToRun.next(false)),
      mergeMap(async val => {
        await callback(val);
        return val;
      }),
      tap(() => freeToRun.next(true)),
    );
  };
}

auditWith is an operator that uses a BehaviorSubject to track whether callback is busy. It uses audit to keep only the most recent value to come down the pipe, and releases that value the next time the BehaviorSubject is marked as free. tap goes before and after the async call in order to make it as busy/free (respectively), and mergeMap is used to actually call the async callback. Note that callback receives the latest value from the pipe, but can return anything it wants (if anything at all); the original value will propagate down the pipe unchanged (by design).