stream-into-iterable.js

//@flow

import type {Stream} from 'most'

function intoAsync<T>(str: Stream<T>): $AsyncIterable<T, void, any> {
  let isEnd = false
  let ask: Array<Dispose<T>> = []
  let tell: Array<T> = []
  const end = str.observe(t => {
    if (ask.length > 0) {
      const dispose = ask[0]
      ask = ask.slice(1)
      dispose.disposed(t)
    } else {
      tell.push(t)
    }
  })
  end.then(() => isEnd = true)
  return async function* Runner(): $AsyncIterable<T, void, any> {
    while (!isEnd) {
      if (tell.length > 0) {
        const value = tell[0]
        tell = tell.slice(1)
        yield value
      } else {
        const dispose: Dispose<T> = new Dispose
        ask.push(dispose)
        const result = await dispose.done
        yield result
      }
    }
  }()
}