A function that takes a stream of events and only emits distinct ones after a debounce timer.

debounceDistinctUntilKeyChanged.spec.ts

import { asyncScheduler, scheduled } from 'rxjs'

import { debounceDistinctUntilKeyChanged } from '.'

describe('debounceDistinctUntilKeyChanged', () => {
  it('should emit as soon as a distinct key appears', () => {
    type TestEvent = { fieldName?: string; fieldValue?: string }

    const source = scheduled<TestEvent>(
      [
        { fieldName: 'name', fieldValue: 'a' },
        { fieldName: 'name', fieldValue: 'ab' },
        { fieldName: 'name', fieldValue: 'abc' },
        { fieldName: 'name', fieldValue: 'abcd' },
        { fieldName: 'title', fieldValue: '1' }
      ],
      asyncScheduler
    )
    const result = source.pipe(debounceDistinctUntilKeyChanged('fieldName', 1000))

    result.subscribe(value => {
      expect(value).toEqual({ fieldName: 'name', fieldValue: 'abcd' })
    })

    // Wait for the debounce time to pass
    setTimeout(() => {
      expect(result).toEqual({ fieldName: 'title', fieldValue: '1' })
    }, 1000)
  })
})

debounceDistinctUntilKeyChanged.ts

/**
 * Watches a stream of events and emits if the key has changed or if the debounce time has passed.
 * based on https://stackoverflow.com/a/66257426/16959
 * @param key key to be used for distinctUntilKeyChanged
 * @param debounceTimeMs time in ms to debounce the action
 */
export function debounceDistinctUntilKeyChanged<T, TKey extends keyof T>(
  key: TKey,
  debounceTimeMs: number
) {
  return (source: Observable<T>): Observable<T> => {
    return new Observable(subscriber => {
      let hasValue = false
      let lastValue: T | undefined = undefined
      let lastKeyValue: T[TKey] | undefined = undefined
      let durationSub: Subscription | undefined = undefined

      const emit = () => {
        durationSub?.unsubscribe()
        durationSub = undefined
        if (hasValue) {
          hasValue = false
          const value = lastValue
          lastValue = undefined
          lastKeyValue = undefined
          subscriber.next(value)
        }
      }

      return source.subscribe({
        next(value) {
          durationSub?.unsubscribe()
          if (hasValue && value[key] !== lastKeyValue) {
            subscriber.next(lastValue)
          }
          hasValue = true
          lastValue = value
          lastKeyValue = value[key]
          durationSub = timer(debounceTimeMs).subscribe(() => {
            emit()
          })
        },
        error(err) {
          subscriber.error(err)
        },
        complete() {
          emit()
          subscriber.complete()
        }
      })
    })
  }
}

Just learned that this can be replaced with

groupBy(object => object.fieldName),
mergeMap(group => group.pipe(debounceTime(500))),