The Missing Guide on Node.js Stream Part 2: Object mode, Read, Write, Transform and Compatibility

Object Mode

Normally stream produce Buffer to the function. To turn on explicitly the object mode in Readable , Transform and Writable you may include.

objectMode: true

Note that each mode is NOT compatible with each other

See guide below to convert from object mode to other

Implement Readable, Transform and Writable

Now in this mode, we will demo how to implement basic from Readable, Transform and Writable.

Big note here is
Callback function callback MUST BE CALLED TO CONTINUE

import { Readable, Transform, Writable } from "stream";
import { pipeline } from "stream/promises";
const readable = new Readable({
  objectMode: true,
  construct(callback) {
    // YOUR VARIABLE HERE
    this.i = 0;
    this.objs = [
      { a: 1, b: 1 },
      { a: 2, b: 2 },
      { a: 3, b: 3 },
    ];
    callback();
  },
  read() {
    console.log("read", this.i);
    this.push(this.objs[this.i++] || null);
  },
});
const transform = new Transform({
  objectMode: true,
  transform(chunk, encoding, callback) {
    try {
      // callback this must be call to continue
      console.log("transform", chunk);
      callback(null, chunk);
    } catch (error) {
      callback(error);
    }
  },
});
const writable = new Writable({
  objectMode: true,
  write(chunk, encoding, callback) {
    // callback this must be call to continue
    console.log("write", chunk);
    callback(null);
  },
});
pipeline(readable, transform, writable)

Not that if you are not familiar with shorthand two below is the same

writable(chunk, encoding, callback) {}
writable: function(chunk, encoding, callback) {}
// SAME

Bridging the Object Mode

Note that below code is not possible.

Because fs consume stream as non-object mode it need Buffer or ‘Byte’

const dest = fs.createWriteStream('dest.json')
readable.pipe(transform).pipe(dest);

/*
TypeError [ERR_INVALID_ARG_TYPE]:
The "chunk" argument must be of type string or
an instance of Buffer or Uint8Array. Received an instance of Object
*/

However slight modify of transform to produce Buffer as output make this work.

const transform = new Transform({
  objectMode: true,
  transform(chunk, encoding, callback) {
    try {
      const buff = Buffer.from(JSON.stringify(chunk));
      console.log("transform", buff)
      callback(null, buff);
    } catch (error) {
      callback(error);
    }
  },
});

const dest = fs.createWriteStream('dest.json')
readable.pipe(transform).pipe(dest);

Hope this help !