an-example.js

var fs        = require('fs');
var pp        = require('path');

var Readable  = require('stream').Readable;
var Writable  = require('stream').Writable;
var Transform = require('stream').Transform;
var util      = require('util');

// divert strem objects by their type
var Switch    = require('./switch' ).Switch;

// writes chunks to console.log
var Log       = require('./log'    ).Log;

// stream that emits File/Directory objects
var Directory = require('./fs'     ).Directory;

// file that streams contents
var File      = require('./fs'     ).File;

// admits a series of readables treams,
// and concats their output to a single readable stream
var Flatten   = require('./flat'   ).Flatten;

var sw = new Switch();
var lg = new Log();

// feed directoy nodes back into switch
sw.case(Directory).pipe( new Flatten() ).pipe( sw );

// write out file contents to log
sw.case(File     ).pipe( new Flatten() ).pipe( lg );

sw.write( new Directory(process.argv[2]) );

flatten.js

var fs        = require('fs');
var pp        = require('path');

var Readable  = require('stream').Readable;
var Writable  = require('stream').Writable;
var Transform = require('stream').Transform;
var util      = require('util');

//----------------

util.inherits(Flatten, Transform);
function Flatten(){
  Transform.call(this,{
    objectMode: true
  });
}

Flatten.prototype._transform = function(readable, _, next){
  if(! readable instanceof Readable) return this.emit('error','Can only flatten Readable objects');
  // console.log("Flattening: %s", readable)
  var self = this;
  var data;
  function read() {
    setImmediate( function(){
      while( (data = readable.read()) && self.push( data ) ) true;
    });
  }
  readable.on('readable', read);
  readable.on('end', next);
  read();
}

exports.Flatten = Flatten;