createAutoCompleteSubscription.ts

/**
 * Setups an autocomplete subscription
 * 
 * @template T
 * @param {Observable<string>} input$ stream of autocomplete-values
 * @param {HTMLInputElement} input the input source
 * @param {Element} startBtn the element to start a query manually
 * @param {Element} cancelBtn the element to end any query
 * @param {(value: T) => void} notifier that there is a definitive value
 * @param {(q: string) => Observable<T>} doQuery returns observable doing the query
 * @param {() => void} startQuery notifier that it has started
 * @param {(reason: string) => void} notifier callback that is has ended
 * @param {(q: string) => void} setQuery notifier that the autocomplete-value is being used
 * @returns {Subscription}
 */
function createAutoCompleteSubscription<T extends ArrayLike<T>>(
  input$: Observable<string>,

  input: HTMLInputElement,
  startBtn: Element,
  cancelBtn: Element,
  
  next: (value: T) => void,

  doQuery: (q: string) => Observable<T>,
  startQuery: () => void,
  endQuery: (reason: string) => void,
  setQuery: (q: string) => void): Subscription {
  
  // Stream of events on the 'cancel' button
  const cancel$ = Observable.fromEvent(cancelBtn, 'click')
    .do({ next: endQuery.bind(this, 'cancel-btn') })

  // Stream of events on the 'query' button
  const query$ = Observable.fromEvent(startBtn, 'click')
    .map(e => input.value)

  // Stream of events of queries that are too short
  const notEnoughInput$ = input$
    .filter((text) => text.length < 3)
    .do({ next: endQuery.bind(this, 'not-enough') })

  // Stream of events of queries that are distinct, debounced and long enough
  const enoughInput$ = input$
    .do({next: setQuery})
    .debounceTime(750)
    .filter((text) => text.length > 2)
    .distinctUntilChanged()

  // Stream of events that should start the call
  const start$ = Observable.merge(query$, enoughInput$)
      .do({next: startQuery})

  // Stream of events that should stop an ongoing call
  const stop$ = Observable.merge(cancel$, enoughInput$, notEnoughInput$)

  // Subscribe
  return start$
    // Each time a start is pushed, use a switchMap to unsubscribe any previous still running observable
    // and isolate errors and such. Anythng indented here can fail silently and the subscription will
    // still be valid
    .switchMap((q) => Observable.of(q)
      .flatMap(doQuery)
      .flatMap(a => Observable.from(a))
      // Once the last timezone (max: 5) is processed end the query
      .take(5)
      .do({next: console.log, complete: endQuery.bind(this, 'completed')})
      // Kill the entire identation if a stop event was pushed
      .takeUntil(stop$)
      .catch((error) => {
        console.warn('something went wrong but subscription is still alive')
        return Observable.empty()
      })
    )
    // For anything that comes out of the switchmap: show it
    .subscribe(next, console.error)
}