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);