crazy.js

// Readable readable :: function(Function<chunk> callback, Boolean isClosed)
function createReadArray(list) {
    var i = 0;
    return function readable(callback, isClosed) {
        debugger;
        if (isClosed) return
        if (i === list.length) return callback(null)
        
        callback(list[i++])
    }
}

// Writable logger :: function(Readable readable)
function logger(readable) {
    readable(function recurse(chunk) {
        console.log("chunk", chunk)
        
        if (chunk !== null) readable(recurse)
    })
}

// Duplex :: Writable -> Writable
// Through map :: Function -> Duplex
var map = Through(function (source, lambda) {
    return function readable(callback, isClosed) {
        source(function (chunk) {
            chunk === null ? callback(null) :
                chunk instanceof Error ? callback(chunk) :
                callback(lambda(chunk))
        }, isClosed)
    }
})

// Duplex twice :: Writable -> Writable
var twice = map(function (i) { return i * 2 })
// Duplex square :: Writable -> Writable
var square = map(function (i) { return i * i })

twice(square(logger))(createReadArray([1, 2, 3]))

pipe(createReadArray([1, 2, 3]))
    .pipe(square)
    .pipe(twice)
    (logger)

function pipe(source) {
    var intermediates = []
    flow.pipe = function (duplex) {
        intermediates.push(duplex)
        return flow
    }
    return flow
    
    function flow(destination) {
        intermediates.reduce(function (writable, duplex) {
            return duplex(writable)
        }, destination)(source)
    }
}

function Through(createReader) {
    return function through(options) {
        return function duplex(destination) {
            return function writable(source) {
                destination(createReader(source, options))
            }
        }
    }
}