Stream examples

streams.js

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

var a = new Readable({ objectMode: true })
var b = new Transform({ objectMode: true })
var c = new Writable({ objectMode: true })

a.counter = 0
a._read = function (size) {
  size = this.counter + size

  var that = this

  if (this.counter > 50) {
    setImmediate(this.push.bind(this, null))
    return
  }

  console.log('READ -->', size, this.counter)

  for (; this.counter < size; this.counter ++) {
    (function (count) {
      console.log('PUSHING -->', count)
      setImmediate(function () {
        that.push({
          value: count
        })
      })
    })(this.counter)
  }
}

b._transform = function (chunk, enc, cb) {
  console.log('TRANSFORM -->', chunk)
  chunk.newPro = 'something-' + chunk.value
  this.push(chunk)
  cb()
}

c._write = function (chunk, enc, cb) {
  console.log('_write called', chunk)
  setImmediate(function () {
    console.log('WRITE -->', chunk)
    cb()
  })
}

a.pipe(b).pipe(c)

uppercase.js

var net = require('net')
var through = require('through2')

net.createServer(function (socket) {
  socket.pipe(through(process)).pipe(socket)
}).listen(4242)

function process (chunk, enc, cb) {
  var string = chunk.toString().toUpperCase()
  console.dir(string)
  this.push(string, 'utf8')
  cb()
}