index.js

/*! multistream. MIT License. Feross Aboukhadijeh <https://feross.org/opensource> */
const stream = require('readable-stream')
const once = require('once')

function toStreams2Obj (s) {
  return toStreams2(s, { objectMode: true, highWaterMark: 16 })
}

function toStreams2Buf (s) {
  return toStreams2(s)
}

function toStreams2 (s, opts) {
  if (!s || typeof s === 'function' || s._readableState) return s

  const wrap = new stream.Readable(opts).wrap(s)
  if (s.destroy) {
    wrap.destroy = s.destroy.bind(s)
  }
  return wrap
}

class MultiStream extends stream.Readable {
  constructor (streams, opts) {
    super({ ...opts, autoDestroy: true })

    this._drained = false
    this._forwarding = false
    this._current = null
    this._toStreams2 = (opts && opts.objectMode) ? toStreams2Obj : toStreams2Buf

    if (typeof streams === 'function') {
      this._queue = streams
    } else {
      this._queue = streams.map(this._toStreams2)
      this._queue.forEach(stream => {
        if (typeof stream !== 'function') this._attachErrorListener(stream)
      })
    }

    this._next()
  }

  _read () {
    this._drained = true
    this._forward()
  }

  _forward () {
    if (this._forwarding || !this._drained || !this._current) return
    this._forwarding = true

    let chunk
    while (this._drained && (chunk = this._current.read()) !== null) {
      this._drained = this.push(chunk)
    }

    this._forwarding = false
  }

  _destroy (err, cb) {
    let streams = []
    if (this._current) streams.push(this._current)
    if (typeof this._queue !== 'function') streams = streams.concat(this._queue)

    if (streams.length === 0) {
      cb(err)
    } else {
      let counter = streams.length
      let er = err
      streams.forEach(stream => {
        destroy(stream, err, err => {
          er = er || err
          if (--counter === 0) {
            cb(er)
          }
        })
      })
    }
  }

  _next () {
    this._current = null

    if (typeof this._queue === 'function') {
      this._queue((err, stream) => {
        if (err) return this.destroy(err)
        stream = this._toStreams2(stream)
        this._attachErrorListener(stream)
        this._gotNextStream(stream)
      })
    } else {
      let stream = this._queue.shift()
      if (typeof stream === 'function') {
        stream = this._toStreams2(stream())
        this._attachErrorListener(stream)
      }
      this._gotNextStream(stream)
    }
  }

  _gotNextStream (stream) {
    if (!stream) {
      this.push(null)
      return
    }

    this._current = stream
    this._forward()

    const onReadable = () => {
      this._forward()
    }

    const onClose = () => {
      if (!stream._readableState.ended && !stream.destroyed) {
        const err = new Error('ERR_STREAM_PREMATURE_CLOSE')
        err.code = 'ERR_STREAM_PREMATURE_CLOSE'
        this.destroy(err)
      }
    }

    const onEnd = () => {
      this._current = null
      stream.removeListener('readable', onReadable)
      stream.removeListener('end', onEnd)
      stream.removeListener('close', onClose)
      stream.destroy()
      this._next()
    }

    stream.on('readable', onReadable)
    stream.once('end', onEnd)
    stream.once('close', onClose)
  }

  _attachErrorListener (stream) {
    if (!stream) return

    const onError = (err) => {
      stream.removeListener('error', onError)
      this.destroy(err)
    }

    stream.once('error', onError)
  }
}

MultiStream.obj = streams => (
  new MultiStream(streams, { objectMode: true, highWaterMark: 16 })
)

module.exports = MultiStream

// Normalize stream destroy w/ callback.
function destroy (stream, err, cb) {
  if (!stream.destroy || stream.destroyed) {
    cb(err)
  } else {
    const callback = once(er => cb(er || err))
    stream
      .on('error', callback)
      .on('close', () => callback())
      .destroy(err, callback)
  }
}