var from = require('from');
var streamworks = require('streamworks');
 
 
// a merge is a stream that splits the input across each function and merges the output back into one stream
var mergestream = streamworks.merge([
 
    // order lots
    function(chunk, enc, callback){
        this.push(chunk + ':10')
        callback()
    },
 
    // order some
    function(chunk, enc, callback){
        this.push(chunk + ':2')
        callback()
    }
])
 
 
// a pipe is a stream that passes values through each function
var pipestream = streamworks.pipe([
 
    // filter anything that does not start with p
    function(chunk, enc, callback){
        if(chunk.toString().indexOf('p')!=0){
            this.push(chunk);
        }
        callback();
    },
 
    // uppercase to input
    function(chunk, enc, callback){
 
        this.push(chunk.toString().toUpperCase())
    }
])
 
// run some data through the merge stream (which will duplicate it) and then through the pipe stream (which will filter it)
from(['apple', 'pie', 'custard'])
    .pipe(mergestream)
    .pipe(pipestream)
    .pipe(process.stdout)

 
var bigAssStream = streamworks.pipe([
  function(chunk, enc, callback){
    if(chunk!='world'){
      this.push(chunk);
    }
    callback();
  },
 
  streamworks.merge([
    function(chunk, enc, callback){
      this.push('A1:' + chunk);
      callback();
    },
    function(chunk, enc, callback){
      this.push('A2:' + chunk);
      callback();
    },
    function(chunk, enc, callback){
      this.push('A3:' + chunk);
      callback();
    }
  ]),
 
  streamworks.pipe([
 
    function(chunk, enc, callback){
      if(chunk.toString().indexOf('A2')!=0){
        this.push(chunk);
      }
      callback();
    },
    streamworks.merge([
      function(chunk, enc, callback){
        this.push('sub1:' + chunk);
        callback(); 
      },
      function(chunk, enc, callback){
        this.push('sub2:' + chunk);
        callback(); 
      }
    ]),
    function(chunk, enc, callback){
      if(chunk.toString().indexOf('sub2:A1:')!=0){
        this.push(chunk);
      }
      
      callback()
    },
  ])
])
 
var arr = [];
 
from(['hello','world','apple']).pipe(bigAssStream)
.on('data', function(chunk){
  arr.push(chunk.toString())
}).on('end', function(){
    console.dir(arr);
})
 
/*
 
sub1:A1:hello
sub1:A3:hello
sub2:A3:hello
sub1:A1:apple
sub1:A3:apple
sub2:A3:apple
    
*/
 

{
    objectMode:true
}