multi-readable.js

/* MultiRead returns a readable that can be sinked
    into multiple writables at the same time

    The way this works is that the readable has to opened and
        closed manually

*/
function MultiRead(readable) {
    var dests = []
    var currentlyReading = false
    var counter = 0
    var currentChunk = null
    var currentPromise = null

    return {
        read: function read() {
            // if we have no value yet then read it
            if (!currentlyReading) {
                currentlyReading = true
                readable.read().then(function (chunk) {
                    // when we have a chunk set it in currentChunk
                    currentChunk = chunk
                    // then check whether all opened readers have asked for it
                    checkReadyToDeliver()
                }, function (error) {
                    currentPromise.reject(error)
                })
            }

            // if we dont have a promise yet create it
            if (!currentPromise) {
                currentPromise = Promise()
            }

            // read was called, increment counter
            counter++

            // check to see whether we should resolve
            checkReadyToDeliver()

            return currentPromise
        },
        abort: readable.abort,
        openParaRead: function (writable) {
            if (dests.indexOf(writable) === -1) {
                dests.push(writable)
            }
        }
    }

    function checkReadyToDeliver() {
        if (currentChunk && counter === dests.length) {
            currentPromise.resolve(currentChunk)

            currentChunk = null
            counter = 0
            currentPromise = null
            currentlyReading = false
        }
    }
}



function sink(writable, readable) {
    if (readable.openParaRead) {
        readable.open(writable)
    }

    pump()
    return readable

    function pump() {
        var chunk = readable.read()

        chunk.then(function (value) {
            if (value === BaseReadableStream.EOF) {
                return writable.close()
            }

            var written = writable.write(value)

            written.then(pump, function (reason) {
                readable.abort(reason)
            })
        }, function (reason) {
            writable.dispose(reason)
        })
    }
}