stream talk

killthatthing.js

/**
 * Readable Stream
 */
require('util').inherits( _rs,  require('stream').Readable  );
function _rs ( fn ) {
	this._read = function(){};
	require('stream').Readable.call( this, { objectMode: true } );
}
function rs ( fn ) {
	return new _rs( fn );
}

/**
 * Writable Stream
 */
require('util').inherits( _ws,  require('stream').Writable  );
function _ws ( fn ) {
	this._write = fn;
	require('stream').Writable.call( this, { objectMode: true } );
}
function ws ( fn ) {
	return new _ws( fn );
}

var n = 0, m = 0
var w = ws(function(data, enc, next){
  data.version += '.' + Math.floor(Math.random() * 10)
  require('fs').writeFileSync('pkgs/'+Date.now()+'.' + (++n) +'.json', JSON.stringify(data))
  m++
  if (m === 50){
    console.log('now we are actually done')
  }
  next()
})


var r = rs()
var i = 50

w.on('thinggotkilled', function(){
  console.log('seems like the stream ended')
})

r.on('killthatthing', function(){
  r.push(null)
  w.emit('thinggotkilled')
})



while(i--){
  r.push(JSON.parse(require('fs').readFileSync('package.json', 'utf8')))
  console.log('another push')
}
r.emit('killthatthing')


console.log('done reading while')

r.on('end', function(){
  console.log('stream ended')
})

setTimeout(function(){
  r.pipe(w)
}, 30)

setTimeout(function(){
  console.log('more poop')
}, 25)


//var i = 1000
//while(i--)