rx-debug.js

Rx.Observable.prototype.debug = function(name = 'default') {
  const s = (window.__subscriptions = window.__subscriptions || {}) // Subscriptions by name.
  s[name] = s[name] || [] // Subscriptions. Useful to track memory leaks.
  let i = -1
  let j = 0
  return Rx.Observable.create(observer => {
    i += 1 // Subscription index.
    j += 1 // Number of live subscriptions.
    const log = (...args) => console.log(`[${name}][${i}] %c%s`, ...args)
    log('color: blue; font-weight: bold', 'subscribe', `(${j} live)`)
    const subscription = this.subscribe({
      next(v) {
        log('', 'next', v)
        observer.next(v)
      },
      complete() {
        log('color: green', 'complete')
        observer.complete()
      },
      error(e) {
        log('color: red', 'error', e)
        observer.error(e)
      },
    })
    subscription.__stack = new Error().stack
    s[name][i] = subscription
    return () => {
      j -= 1
      log('color: blue; font-weight: bold', 'unsubscribe', `(${j} live)`)
      subscription.unsubscribe()
    }
  })
}

Rx.Observable.prototype._withLatestFrom = function(...args) {
  const debuggedArgs = args.map((arg, i) => (arg.subscribe ? arg.debug(`withLatestFrom ${i}`) : arg))
  return this.withLatestFrom(...debuggedArgs)
}

Rx.Observable.prototype._withLatestFrom = function(...args) {
  const emitted = []
  const spiedArgs = args.map((arg, index) => {
    if (!(arg instanceof Rx.Observable)) return arg
    emitted[index] = false
    return arg.do(() => (emitted[index] = true))
  })
  let tid
  return this.do(() => {
    if (tid) return
    tid = setTimeout(() => {
      const missing = emitted.reduce(
        (indices, bool, boolIndex) => (bool === false ? [...indices, boolIndex] : indices),
        [],
      )
      if (missing.length > 0) console.log(`Did not emit: ${missing.join(', ')}`)
    }, 1000)
  }).withLatestFrom(...spiedArgs)
}