lib/pipeStream.js
function addFunctions(self){
    // Function for assigning a process to run after the end of this PipeStream
    self.and = function(){
        self._and = new ShellStream(arguments);
        self.on('exit',function(code){
            if(code !== 0){
                self.emit('error',code);
                return;
            }
            self._and = self._and.run();
        });
        return self._and;
    };
  
    var oldPipe = self.pipe ? self.pipe.bind(self) : false;
    // Function for adding a new pipe to the pipeline
    self.pipe = function(){
        // Create pipe and add it to the pipeline hierarchy
        self._pipe = createPipeStream.apply({}, arguments);
        self._pipe._parent = self;
        self._addPipe(self._pipe);
        self._pipe._pipeline = self._pipeline;

        // Pipe the necessary data events to the new PipeStream
        if(utils.isSpawn(self)){
            self.stdout.pipe(utils.isSpawn(self._pipe) ? self._pipe.stdin:  self._pipe);
            self.stderr.on('data', function(d){
                self._pipe.emit('error', d);
            });
        } else {
            if(utils.isSpawn(self._pipe)){
                oldPipe(self._pipe.stdin);
                self.on('error',function(d){
                    self._pipe.stderr.emit('data', d);
                });
            } else {
                oldPipe(self._pipe);
                self.on('error',function(d){
                    self._pipe.emit('error', d);
                });
            }
        }
        // return new PipeStream
        return self._pipe;
    };

    // Internal function for appending a PipeStream to pipeline
    Object.defineProperty(self, '_addPipe',{value:function(pipe){
        self._pipeline.push(pipe);
        pipe._nr = self._pipeline.length - 1;
        pipe._error = self._error;
    }});


    // Internal function for destroying the whole pipeline
    Object.defineProperty(self, '_cleanUp', {
        value: function(){
            this._pipeline.forEach(function(p){
                p._error = null; // Make sure error is called only once.
                if(p.kill){
                    p.kill();
                } else if(p.destroy){
                    p.destroy();
                }
            });
        }
    });

    // Function for retrieving the pipeline beginning
    self.first = function(){
        return self.get(0);
    };

    // Function for retrieving the pipeline end
    self.last = function(){
        return self.get(self._pipeline.length -1);
    };

    // Function for retrieving the nth element in pipeline
    self.get = function(n){
        return self._pipeline[n] || false;
    };

    // Function for appending an error handler for all elements in pipeline
    self.error = function(fn){
        self._pipeline.forEach(function (section) {
            section._error = fn;
        });
        return self;
    };
}
function addFunctions(self){
    // Function for assigning a process to run after the end of this PipeStream
    self.and = function(){
        self._and = new ShellStream(arguments);
        self.on('exit',function(code){
            if(code !== 0){
                self.emit('error',code);
                return;
            }
            self._and = self._and.run();
        });
        return self._and;
    };
  
    var oldPipe = self.pipe ? self.pipe.bind(self) : false;
    // Function for adding a new pipe to the pipeline
    self.pipe = function(){
        // Create pipe and add it to the pipeline hierarchy
        self._pipe = createPipeStream.apply({}, arguments);
        self._pipe._parent = self;
        self._addPipe(self._pipe);
        self._pipe._pipeline = self._pipeline;

        // Pipe the necessary data events to the new PipeStream
        if(utils.isSpawn(self)){
            self.stdout.pipe(utils.isSpawn(self._pipe) ? self._pipe.stdin:  self._pipe);
            self.stderr.on('data', function(d){
                self._pipe.emit('error', d);
            });
        } else {
            if(utils.isSpawn(self._pipe)){
                oldPipe(self._pipe.stdin);
                self.on('error',function(d){
                    self._pipe.stderr.emit('data', d);
                });
            } else {
                oldPipe(self._pipe);
                self.on('error',function(d){
                    self._pipe.emit('error', d);
                });
            }
        }
        // return new PipeStream
        return self._pipe;
    };

    // Internal function for appending a PipeStream to pipeline
    Object.defineProperty(self, '_addPipe',{value:function(pipe){
        self._pipeline.push(pipe);
        pipe._nr = self._pipeline.length - 1;
        pipe._error = self._error;
    }});


    // Internal function for destroying the whole pipeline
    Object.defineProperty(self, '_cleanUp', {
        value: function(){
            this._pipeline.forEach(function(p){
                p._error = null; // Make sure error is called only once.
                if(p.kill){
                    p.kill();
                } else if(p.destroy){
                    p.destroy();
                }
            });
        }
    });

    // Function for retrieving the pipeline beginning
    self.first = function(){
        return self.get(0);
    };

    // Function for retrieving the pipeline end
    self.last = function(){
        return self.get(self._pipeline.length -1);
    };

    // Function for retrieving the nth element in pipeline
    self.get = function(n){
        return self._pipeline[n] || false;
    };

    // Function for appending an error handler for all elements in pipeline
    self.error = function(fn){
        self._pipeline.forEach(function (section) {
            section._error = fn;
        });
        return self;
    };
}
Adds extra functionality to EventEmitter

@param {EventEmitter} self
@returns {EventEmitter}