Create a Transform Stream in Node.js

A transform stream is a type of duplex stream that has the ability to modify the data as it passes through. Transform streams are commonly used for tasks such as data compression, encryption, or formatting.

They are usually connected to a readable and a writable stream through the pipe() method.

readable.pipe(transform).pipe(writable);

To create a transform stream, you can create a class that inherits from the Transform class and override the _transform() method:

const { Transform } = require('stream');

class TransformStream extends Transform {
  constructor(options) {
    super(options);
  }

  _transform(chunk, encoding, callback) {
    // Transform the data chunk

    // Add the chunk to the read queue
    this.push(chunk);

    // Execute the callback function
    callback();
  }
}

Where:

• chunk is a chunk of data being passed through the stream.
• encoding is the character encoding of the data chunk.
• this.push() is a method used to add the processed chunk of data to the read queue
• callback is a function that must be called once the processing of the chunk is complete.

When there is no more written data to be consumed, you can execute a set of "wrap up" instructions before the 'end' event is emitted by the readable stream, by overriding the _flush() method:

const { Transform } = require('stream');

class TransformStream extends Transform {
  constructor(options) {
    super(options);
  }

  _transform(chunk, encoding, callback) {
    // Process chunk
  }

  _flush(callback) {
    // Wrap up task
    callback();
  }
}

Example

📚 Definition: In JavaScript, the bitwise XOR operator ^ is used to compare binary numbers bit by bit and either:

• Set the resulting bit to 0 if both bits equal 0.
• Set the resulting bit to 0 if both bits equal 1.
• Set the resulting bit to 1 if bits are unequal.

For example, the result of the bitwise operation 5 ^ 3 is 6, as demonstrated by the following bit comparison:

5 (base 10)  =  00000101  (base 2)
3 (base 10)  =  00000011  (base 2)
                --------
6 (base 10)  =  00000110  (base 2)

In cryptography, the XOR cipher is a symmetric encryption technique that applies the bitwise XOR operation between each byte of data and a key byte.

Let's consider the following file that contains a list of username and password pairs in cleartext separated by a colon (:) character:

$ cat users.txt
johndoe:9j!dlD-MC5
alice:86KXL(*&/f
bob:33*Msdf@mD

Let's consider this script, that implements a transform stream that encrypts a data stream using the XOR cipher:

const { createReadStream, createWriteStream } = require('node:fs');
const { Transform } = require('node:stream');

class XORTransformStream extends Transform {
  constructor(key, options) {
    // Invoke the Transform stream constructor
    super(options);
    
    // Store the encryption key
    this.key = key;
  }

  _transform(chunk, encoding, callback) {
    // Create a buffer of the size of the data chunk
    let buffer = Buffer.alloc(chunk.length);

    // Perform a XOR between the buffer's bytes and the encryption key
    for (let i = 0; i < chunk.length; i++) {
      buffer[i] = chunk[i] ^ this.key[i % this.key.length];
    }

    // Push the encrypted chunk to the other end of the stream
    this.push(buffer);

    // Complete the processing
    callback();
  }
}

const source = 'users.txt';
const destination = 'users.crypt';
const key = 'SecretEncryptionKey';

const read = createReadStream(source);
const write = createWriteStream(destination);
const transform = new XORTransformStream(Buffer.from(key));

read.pipe(transform).pipe(write);

read.on('error', e => console.error(e.toString()));
write.on('error', e => console.error(e.toString()));
transform.on('error', e => console.error(e.toString()));

When executed, it will:

1. Create a subclass of the Transform class named XORTransformStream that overrides the _transform() method in charge of gathering the data chunks and encrypt them using the XOR cipher.
2. Create an instance of a readable stream that reads from the users.txt file using the createReadStream() method.
3. Create an instance of a writable stream that writes to the users.crypt file using the createWriteStream() method.
4. Create an instance of a transform stream using the XORTransformStream class.
5. Connect the read, transform, and write streams using the pipe() method.

Which will produce this output:

$ cat users.crypt
9

 TZX-BPs2
N}X(*5X^OA1_PAO96

Summary

Here's a summary of what you've learned in this lesson:

• A transform stream is used to transform the data as it is read and written.
• The stream.Transform class is used to create a transform stream that can modify data as it passes through.