Transform & Duplex

import { Transform, PassThrough } from 'node:stream';
import { pipeline } from 'node:stream/promises';
 
const tap = new PassThrough();
tap.on('data', (chunk) => metrics.bytes(chunk.length));
 
await pipeline(source, tap, destination);

import { Transform, PassThrough, Duplex } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { Readable } from 'node:stream';
 
const splitLines = new Transform({
  transform(chunk, _enc, cb) {
    const parts = String(chunk).split('\n');
    const tail = parts.pop() ?? '';
    for (const line of parts) if (line) this.push(line);
    cb(null, tail); // buffer incomplete line in internal state via passing as chunk to next
  },
  flush(cb) {
    cb();
  },
});
 
function tee<T extends NodeJS.ReadableStream>(source: T): [T, PassThrough] {
  const branch = new PassThrough();
  source.pipe(branch);
  return [source, branch];
}
 
// Duplex conceptual - net.Socket extends Duplex
const socketLike = new Duplex({
  read() { /* pull from underlying resource */ },
  write(chunk, _enc, cb) { cb(); /* send to peer */ },
});

import { createGzip } from 'node:zlib';
 
await pipeline(
  Readable.from(['{"a":1}\n', '{"b":2}\n']),
  splitLines,
  createGzip(),
  process.stdout,
);

import { Transform, type TransformCallback } from 'node:stream';
 
function createJsonParser(): Transform {
  let buffer = '';
  return new Transform({
    objectMode: true,
    transform(chunk: Buffer, _enc: BufferEncoding, cb: TransformCallback) {
      buffer += chunk.toString('utf8');
      const lines = buffer.split('\n');
      buffer = lines.pop() ?? '';
      try {
        for (const line of lines) if (line) this.push(JSON.parse(line));
        cb();
      } catch (err) {
        cb(err as Error);
      }
    },
  });
}