drainObservable.js

function * drainObservable (observable) {
  let buf = []
  let waiting = []
  let err = null
  let done = false
  let subscription

  try {
    subscription = observable.subscribe(val => {
      if (waiting.length) waiting.shift().resolve(val)
      else buf.push(Promise.resolve(val))
    }, error => {
      done = true
      if (waiting) waiting.shift().reject(error)
      else err = error
    }, () => {
      done = true
    })

    while (!done) {
      if (err) throw err
      if (buf.length) yield buf.shift()
      else {
        const { resolve, reject, promise } = deferred()
        waiting.push({ resolve, reject })
        yield promise
      }
    }
  } finally {
    if (subscription) subscription.unsubscribe()
  }
}