rxjs-drop-intermediate

RxJS operator to prevent congestion by dropping intermediate values. This can be useful when there are many events generated by user-input or a WebSocket connection where some intermediate events (e.g. updates from a certain channel) can be dropped.

Usage:

import { dropIntermediate } from 'rxjs-drop-intermediate';

const droppingObservable = source.pipe(dropIntermediate());

Example taken from integration tests:

const subject = new Subject<string>();
webSocket.onmessage = (m) => subject.next(m.data.toLocaleString());
const actual = await firstValueFrom(
  subject.pipe(
    dropIntermediate(),
    map((v) => parseInt(v)),
    tap(() => execSync('sleep 2')),
    take(2),
    toArray(),
    map((a) => a[1] - a[0])
  )
);

expect(actual).toBeGreaterThan(5);