A way of using rxjs to wire-up a cancellable, debounced action

debounceUnlessCancelled.ts

import { Observable, race, Subject, SubscribableOrPromise } from 'rxjs'
import { debounce, filter } from 'rxjs/operators'

describe('debounceUnlessCancelled', () => {

  let action: jest.Mock<string>
  let input: Subject<string>
  let cancel: Subject<void>
  let manualTrigger: Subject<string>
  beforeEach(() => {
    action = jest.fn()
    input = new Subject<string>()
    cancel = new Subject<void>()
    manualTrigger = new Subject<string>()

    const trigger = (): Observable<string> => manualTrigger
    debounceUnlessCancelled({ input, cancel, trigger }).subscribe(action)
  })

  it('does not action values immediately', () => {
    input.next('foo')
    expect(action).not.toHaveBeenCalledWith('foo')
  })

  it('action values when the debounce mechanism is triggered', () => {
    input.next('foo')
    manualTrigger.next()
    expect(action).toHaveBeenCalledWith('foo')
  })

  it('actions the last value when the debounce mechanism is triggered', () => {
    input.next('foo')
    input.next('bar')
    manualTrigger.next()
    expect(action).not.toHaveBeenCalledWith('foo')
    expect(action).toHaveBeenCalledWith('bar')
  })

  it('clears the cache on reset', () => {
    input.next('foo')
    input.next('bar')
    cancel.next()
    manualTrigger.next()
    expect(action).not.toHaveBeenCalledWith('bar')
  })

})

interface Props<T> {
  input: Observable<T>
  cancel: Observable<void>
  trigger(value: T): SubscribableOrPromise<void>
}

/**
 * The manual trigger is useful for unit-testing. To use for something that debounces after
 * 200ms unless cancelled use e.g. `trigger = () => timer(200)`
 */
function debounceUnlessCancelled<T>({ input, cancel, trigger }: Props<T>): Observable<T> {
  const resetOrDebounce = race(cancel, input.pipe(debounce(trigger)))
  return resetOrDebounce.pipe(filter(o => !!o)) as Observable<T>
}