import { Readable } from 'stream'

const createReadable = generator => {
  const readable = new Readable()
  let iterator
  readable._read = size => {
    iterator = generator(size)
    readable._destroy = error => {
      iterator.throw(error)
    }
    const read = readable._read = async size => {
      try {
        const cursor = await iterator.next(size)
        if (cursor.done) {
          return readable.push(null)
        }
        readable.push(cursor.value)
      } catch (error) {
        readable.destroy(error)
      }
    }
    return read(size)
  }

  return readable
}

{
  const { close, open, read } = require('fs')
  const { fromCallback } = require('promise-toolbox')

  const stream = createReadable(async function * (size) {
    const fd = await fromCallback(cb => open(__filename, 'r', cb))
    try {
      while (true) {
        const buf = Buffer.allocUnsafe(size)
        const n = await fromCallback(cb => read(fd, buf, 0, size, null, cb))
        if (n < size) {
          yield buf.slice(0, n)
          break
        }
        size = yield buf
      }
    } finally {
      await fromCallback(cb => close(fd, cb))
    }
  })

  stream.pipe(process.stdout)
}