2020-12-12-window-by.ts

export const windowBy = <K, T>(
  keyFn: (t: T) => K
): OperatorFunction<T, GroupedObservable<K, T>> => (source) => source.pipe(
  publish((sharedSource) => {
    const newWindow = sharedSource.pipe(
      distinctUntilChanged((a, b) => keyFn(a) === keyFn(b))
    );
    return sharedSource.pipe(
      groupBy(keyFn, undefined, () =>
        newWindow.pipe(
          // Skip the first window as that is what created the current group.
          skip(1)
        )
      )
    );
  })
);