Transform objects in a Node stream

By default, a Node.js stream expects to operate on a Buffer or a Uint8Array. We can override this by telling the stream to use “object mode”.

Consider the following array of TypeScript objects.

interface Person {
  id: number
  name: string
  age: number
}

const data: Person[] = [
  { id: 1, name: 'John Doe', age: 32 },
  { id: 2, name: 'Jane Doe', age: 28 },
  { id: 3, name: 'Gabe Doe', age: 63 },
]

Our hypothetical stream outputs the objects, one at a time. We want to accept the object, remove the id property, and pass it along.

We achieve this using a Transform stream, configured to read and write objects.

• The writeableObjectMode option lets the stream accept an object.
• The readableObjectMode option lets the stream return an object.

import { Transform, TransformOptions } from 'stream'

class RemoveId extends Transform {
  constructor(opts?: TransformOptions) {
    super({ ...opts, readableObjectMode: true, writeableObjectMode: true })
  }
}

The object mode properties are immutable. This is why we pass them to the parent constructor.

With that in place, we can write the transformation method.

import { Transform, TransformCallback } from 'stream'

interface Person {
  id: number
  name: string
  age: number
}

class RemoveId extends Transform {
  // Constructor omitted for brevity

  _transform(chunk: Person, _: string, callback: TransformCallback) {
    const { _id, ...transformed } = chunk
    this.push(transformed)
    callback()
  }
}

Here’s the complete Transform class.

import { Transform, TransformCallback, TransformOptions } from 'stream'

interface Person {
  id: number
  name: string
  age: number
}

class RemoveId extends Transform {
  constructor(opts?: TransformOptions) {
    super({ ...opts, readableObjectMode: true, writeableObjectMode: true })
  }

  _transform(chunk: Person, _: string, callback: TransformCallback) {
    const { _id, ...transformed } = chunk
    this.push(transformed)
    callback()
  }
}

hypotheticalStream.pipe(new RemoveId())