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;
    };
}
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}
Adds extra functionality to EventEmitter