index.js

const { Duplex } = require('readable-stream')
const endOfStream = require('end-of-stream')
const once = require('once')
const noop = () => {}

const IGNORE_SUBSTREAM = {}


class ObjectMultiplex extends Duplex {

  constructor(_opts = {}) {
    const opts = Object.assign({}, _opts, {
      objectMode: true,
    })
    super(opts)

    this._substreams = {}
  }

  createStream (name) {
    // validate name
    if (!name) throw new Error('ObjectMultiplex - name must not be empty')
    if (this._substreams[name]) throw new Error('ObjectMultiplex - Substream for name "${name}" already exists')

    // create substream
    const substream = new Substream({ parent: this, name: name })
    this._substreams[name] = substream

    // listen for parent stream to end
    anyStreamEnd(this, (err) => {
      substream.destroy(err)
    })

    return substream
  }

  // ignore streams (dont display orphaned data warning)
  ignoreStream (name) {
    // validate name
    if (!name) throw new Error('ObjectMultiplex - name must not be empty')
    if (this._substreams[name]) throw new Error('ObjectMultiplex - Substream for name "${name}" already exists')
    // set
    this._substreams[name] = IGNORE_SUBSTREAM
  }

  // stream plumbing

  _read () {}

  _write(chunk, encoding, callback) {
    // parse message
    const name = chunk.name
    const data = chunk.data
    if (!name) {
      console.warn(`ObjectMultiplex - malformed chunk without name "${chunk}"`)
      return callback()
    }

    // get corresponding substream
    const substream = this._substreams[name]
    if (!substream) {
      console.warn(`ObjectMultiplex - orphaned data for stream "${name}"`)
      return callback()
    }

    // push data into substream
    if (substream !== IGNORE_SUBSTREAM) {
      substream.push(data)
    }

    callback()
  }

}


class Substream extends Duplex {

  constructor ({ parent, name }) {
    super({
      objectMode: true,
    })

    this._parent = parent
    this._name = name
  }

  _read () {}

  _write (chunk, enc, callback) {
    this._parent.push({
      name: this._name,
      data: chunk,
    })
    callback()
  }

}

module.exports = ObjectMultiplex

// util

function anyStreamEnd(stream, _cb) {
  const cb = once(_cb)
  endOfStream(stream, { readable: false }, cb)
  endOfStream(stream, { writable: false }, cb)
}