rxjs: maxCountDuringPeriod

maxCountDuringPeriod.ts

/**
 * maxCountDuringPeriod
 */
export const maxCountDuringPeriod = (count: number, period: number) => (obs: Observable<any>) => {
  if (count < 2) {
    throw new Error('maxCountDuringPeriod count must be greater than or equal to 2');
  }

  if (period < 0) {
    throw new Error('maxCountDuringPeriod period must be positive');
  }

  return obs.pipe(
    map(_ => currentUnixTime()),
    bufferCount(count, 1), // Emit arrays of unix time (by event)
    filter(xs => xs.length === count), // Filter out buffers that don't exceed count restriction
    map(xs => xs[xs.length - 1] - xs[0]), // Map to seconds between first and last event
    filter(x => x <= period), // Filter out groupings that occur outside of the period restraint
  );
};