Creating object mode transform streams

If our stream is not returning serializable data (a buffer or a string), we need to make it use object mode. Object mode just means that the values returned are generic objects and the only difference is how much data is buffered. Per default, when not using object mode, the stream will buffer around 16 KB of data before pausing. When using object mode, it will start pausing when 16 objects have been buffered.

Let's create a folder called object-streams, initialize it as a package, install through2 and ndjson, and create an index.js file:

$ mkdir object-streams 
$ cd object-streams 
$ npm init -y  
$ npm install through2 ndjson 
$ touch index.js 

Let's make index.js look like this:

const through = require('through2') 
const { serialize } = require('ndjson') 
 
const xyz = through.obj(({x, y}, enc, cb) => { 
  cb(null, {z: x + y}) 
}) 
 
xyz.pipe(serialize()).pipe(process.stdout) 
 
xyz.write({x: 199, y: 3}) 
 
xyz.write({x: 10, y: 12}) 

This will output the following:

{"z":202} 
{"z":22} 

We can create an object stream with through2 using the obj method. The behavior of through.obj is the same as through, except instead of data chunks, our transform function receives and responds with objects.

We use the ndjson module's serialize function to create a serializer stream which converts streamed objects into newline delimited JSON. The serializer stream is a hybrid stream where the writable side is in object mode, but the readable side isn't. Objects go in; buffers come out.

With core streams, we pass an objectMode option to create an object stream instead.

Let's create a core.js file in the same folder, and install the readable-stream module:

$ touch core.js 
$ npm install --save readable-stream 

Now we'll fill it with the following code:

const { Transform } = require('readable-stream') 
const { serialize } = require('ndjson') 
 
const xyz = Transform({ 
  objectMode: true, 
  transform: ({x, y}, enc, cb) => { cb(null, {z: x + y}) }  
}) 
 
xyz.pipe(serialize()).pipe(process.stdout) 
 
xyz.write({x: 199, y: 3}) 
 
xyz.write({x: 10, y: 12}) 

As before ,we'll see the following output:

{"z":202} 
{"z":22}