object streams

Normally you can only read and write buffers and strings with streams. However, if you initialize a stream in objectMode, you can use any kind of object (except for null).

this lets you parse objects. you can have a whole pipeline with many steps, without having to serialise and deserialise a bunch of times. Can be useful for optimisations.

var through = require('through2')
// this will initialise it (through.obj) ↴
var tr = through.obj(function (row, enc, next) {
  next(null, (row.n * 1000) + '\n')
})
tr.pipe(process.stdout)
tr.write({ n: 5 })
tr.write({ n: 10 })
tr.write({ n: 3 })
tr.end()

Of course, the destination has to be able to receive it as well.

---

output:

5000
10000
3000

Let's do an example:

  //	****		obj.js		****

var through = require('through2')
// **** 3. keep the number of bytes seen  ↴
var size = 0
// read data from stdin ↴
process.stdin
  .pipe(through.obj(write1))
  // or you could do (longer version):
  // .pipe(through({ objectMode: true }, write))
  .pipe(through.obj(write2))

function write1 (buf, enc, next) {
  // **** 4. put this out to the stream  ↴
  next(null, { length: buf.length, total: size += buf.length })
}

// here, we get Object as buffer ↴
function write2 (obj, enc, next) {
  console.log('obj=', obj)
  next()
}
function end () {
  console.log('size=', size)
}

This will produce $ node obj.js: obj = { length: 4, total: 8 }

Then use the through module. The core module can take a function flush or with through2 and end

  //	****		obj.js		****

var through = require('through2')
var size = 0
process.stdin
  .pipe(through.obj(write1))
  // **** 1. at the `end` callback  ↴
  .pipe(through.obj(write2, end))

function write1 (buf, enc, next) {
  // **** 3. move the increment to write2  ↴
  next(null, { length: buf.length })
}

function write2 (obj, enc, next) {
  // **** 4. increment goes here, logging to `end`  ↴
  size += obj.length
  next()
}
// **** 2. now we can do something with the end  ↴
function end () {
  console.log('size=', size)
}

Later we will use modules that make using objects with streams easier.

Now, the total number of bytes will be logged, when we terminate the stdin (CTRL+d).

You only get the end event if you read from the stream

If you don't hook up a read from the stream you will never get an end event.