splitter.js

var cadence = require('cadence')

var Procession = require('./procession')

function Splitter (queue, splits) {
    this._shifter = queue.shifter()
    this._map = {}
    this._array = []
    for (var key in splits) {
        var queue = new Procession
        var split = {
            selector: splits[key],
            queue: queue,
            shifter: queue.shifter()
        }
        this._map[key] = split
        this._array.push(split)
    }
}

Splitter.prototype.dequeue = cadence(function (async, name) {
    var split = this._map[name]
    var loop = async(function () {
        if (split.shifter.peek() != null || split.queue.endOfStream) {
            return [ loop.break, split.shifter.shift() ]
        }
        this._shifter.dequeue(async())
    }, function (envelope) {
        if (envelope == null) {
            this._array.forEach(function (split) { split.queue.push(null) })
        } else {
            for (var i = 0, I = this._array.length; i < I; i++) {
                if (this._array[i].selector(envelope)) {
                    this._array[i].queue.push(envelope)
                    break
                }
            }
        }
    })()
})

module.exports = Splitter