中文
devexamples

Node Stream Transform Chunk Pipeline

Chain readable, transform, and writable streams with pipeline to process large files in constant memory while propagating backpressure and errors.

JavascriptNodeFile Upload

Code

snippet
import { createReadStream, createWriteStream } from 'node:fs';
import { Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { createGzip } from 'node:zlib';

class LineCount extends Transform {
  constructor() {
    super({ objectMode: false });
    this.lines = 0;
  }

  _transform(chunk, encoding, callback) {
    const text = chunk.toString('utf8');
    this.lines += (text.match(/\n/g) ?? []).length;
    this.push(text.toUpperCase());
    callback();
  }

  _flush(callback) {
    this.push('\n# total lines: ' + this.lines + '\n');
    callback();
  }
}

await pipeline(
  createReadStream('input.log', { encoding: 'utf8', highWaterMark: 64 * 1024 }),
  new LineCount(),
  createGzip({ level: 6 }),
  createWriteStream('output.log.gz'),
);

console.log('pipeline finished');

Related Examples