pump.js

var PumpLink = exports.PumpLink = function (readStream, writeStream, cb) {
    var self = this;
    this._rs = readStream
    , this._ws = writeStream
    , this._endcb;
    
    this._datacb = function (chunk) {
        if (writeStream.write(chunk) === false) {
            var l = function () {
                writeStream.removeListener("drain", l);
                readStream.resume();
            }
            writeStream.addListener("drain", l);
            readStream.pause();
        }
    }

    readStream.addListener("data", this._datacb);

    if(cb) {
        this._endcb = function () {
            cb(self._rs, self._ws);
        }
        readStream.addListener("end", this._endcb);
    }
}

PumpLink.prototype.unlink = function() {
    this._rs.removeListener("data", this._datacb);
    if(this._endcb) this._rs.removeListener("end", this._endcb);
}

exports.pump = function (readStream, writeStream, ecb) {
    return new PumpLink(readStream, writeStream, ecb);
}