Implementing Custom Streams

import { Readable, Writable } from 'node:stream';

// Subclass style: private state in fields, methods with the underscore.
class Hello extends Readable {
  _read() {
    this.push('hello');
    this.push(null);
  }
}

// Simplified construction: the same methods, passed as options.
const sink = new Writable({
  write(chunk, encoding, callback) {
    console.log('got', chunk.toString());
    callback();
  },
});

new Hello().pipe(sink);

import { Readable } from 'node:stream';

class Countdown extends Readable {
  #next;
  constructor(from, options) {
    super({ ...options, objectMode: true });
    this.#next = from;
  }
  _read() {
    // Push until push() returns false or the source is exhausted.
    let more = true;
    while (more) {
      if (this.#next < 0) {
        this.push(null); // end of stream
        return;
      }
      more = this.push({ n: this.#next-- });
    }
  }
}

const values = [];
for await (const item of new Countdown(5)) values.push(item.n);
console.log(values);

[ 5, 4, 3, 2, 1, 0 ]

import { Readable } from 'node:stream';
import { setTimeout as sleep } from 'node:timers/promises';

// Stand-in for a paged API or a database cursor.
let page = 0;
const fetchPage = async (n) => {
  await sleep(5);
  return n < 3 ? [`row-${n}-a`, `row-${n}-b`] : [];
};

const paged = new Readable({
  objectMode: true,
  read() {
    fetchPage(page++).then(
      (rows) => {
        if (rows.length === 0) this.push(null);
        else for (const r of rows) this.push(r);
      },
      (err) => this.destroy(err), // errors go through destroy(), never throw
    );
  },
});

console.log(await paged.toArray());

[ 'row-0-a', 'row-0-b', 'row-1-a', 'row-1-b', 'row-2-a', 'row-2-b' ]

import { Readable, Writable } from 'node:stream';
import { pipeline } from 'node:stream/promises';

class BatchWriter extends Writable {
  #batch = [];
  #size;
  #sink;
  constructor(sink, { batchSize = 3 } = {}) {
    super({ objectMode: true, highWaterMark: 4 });
    this.#size = batchSize;
    this.#sink = sink;
  }
  _write(row, _encoding, callback) {
    this.#batch.push(row);
    if (this.#batch.length < this.#size) return callback();
    this.#flush().then(() => callback(), callback);
  }
  _final(callback) {
    // Called after the last write(), before 'finish'.
    this.#flush().then(() => callback(), callback);
  }
  async #flush() {
    if (this.#batch.length === 0) return;
    const rows = this.#batch;
    this.#batch = [];
    await this.#sink(rows);
  }
}

const batches = [];
await pipeline(
  Readable.from(Array.from({ length: 8 }, (_, i) => ({ id: i }))),
  new BatchWriter(async (rows) => { batches.push(rows.map((r) => r.id)); }),
);
console.log(batches);

[ [ 0, 1, 2 ], [ 3, 4, 5 ], [ 6, 7 ] ]

import { Writable } from 'node:stream';

const w = new Writable({
  objectMode: true,
  write(chunk, _enc, cb) { console.log('write', chunk); setImmediate(cb); },
  writev(entries, cb) { console.log('writev', entries.map((e) => e.chunk)); setImmediate(cb); },
});

w.cork();
w.write('a'); w.write('b'); w.write('c');
w.uncork();
w.end('d');
await new Promise((resolve) => w.on('finish', resolve));

writev [ 'a', 'b', 'c' ]
write d

import { Duplex } from 'node:stream';

class Echo extends Duplex {
  #resume = null;
  constructor() {
    super({ objectMode: true, readableHighWaterMark: 2 });
  }
  _write(message, _encoding, callback) {
    // Bridge backpressure: hold the write callback while the readable side is full.
    if (this.push(`echo: ${message}`)) callback();
    else this.#resume = callback;
  }
  _read() {
    const resume = this.#resume;
    this.#resume = null;
    resume?.();
  }
  _final(callback) {
    this.push(null);
    callback();
  }
}

const echo = new Echo();
for (const word of ['one', 'two', 'three', 'four', 'five']) echo.write(word);
echo.end();
console.log(await echo.toArray());

[ 'echo: one', 'echo: two', 'echo: three', 'echo: four', 'echo: five' ]

import { Readable, Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { StringDecoder } from 'node:string_decoder';

// Bytes in, row arrays out.
export class CsvParser extends Transform {
  #decoder = new StringDecoder('utf8');
  #rest = '';
  constructor() {
    super({ readableObjectMode: true });
  }
  _transform(chunk, _encoding, callback) {
    this.#rest += this.#decoder.write(chunk); // safe for characters split across chunks
    const lines = this.#rest.split(/\r?\n/);
    this.#rest = lines.pop(); // the last element is an unterminated line
    for (const line of lines) if (line) this.push(line.split(','));
    callback();
  }
  _flush(callback) {
    this.#rest += this.#decoder.end();
    if (this.#rest) this.push(this.#rest.split(','));
    callback();
  }
}

// Split the euro sign (three bytes) across two chunks on purpose.
const bytes = Buffer.from('title,price\nPraga,10€\nDune,8\n');
const cut = bytes.indexOf(0xe2) + 1;

const rows = [];
await pipeline(
  Readable.from([bytes.subarray(0, cut), bytes.subarray(cut)], { objectMode: false }),
  new CsvParser(),
  async function (source) { for await (const row of source) rows.push(row); },
);
console.log(rows);

[ [ 'title', 'price' ], [ 'Praga', '10€' ], [ 'Dune', '8' ] ]

import { Transform } from 'node:stream';

const upper = new Transform({
  transform(chunk, _encoding, callback) {
    callback(null, chunk.toString().toUpperCase());
  },
});

upper.end('abc');
upper.on('data', (chunk) => console.log(String(chunk)));

ABC

import { Readable } from 'node:stream';
import { pipeline } from 'node:stream/promises';

async function* books() {
  yield { title: 'Dune', year: 1965 };
  yield { title: 'Neuromancer', year: 1984 };
  yield { title: 'Hyperion', year: 1989 };
}

const out = [];
await pipeline(
  Readable.from(books()),                                 // objectMode is true by default
  async function* (source) {                              // generator as a transform stage
    for await (const b of source) {
      if (b.year > 1970) yield { ...b, title: b.title.toUpperCase() };
    }
  },
  async function (source) {                               // async function as the last stage (sink)
    for await (const b of source) out.push(b);
  },
);
console.log(out);

// Strings and Buffers are NOT iterated by Readable.from()
console.log(await Readable.from('abc').toArray());
console.log(await Readable.from('abc', { objectMode: false }).toArray());

[
  { title: 'NEUROMANCER', year: 1984 },
  { title: 'HYPERION', year: 1989 }
]
[ 'abc' ]
[ <Buffer 61 62 63> ]

import { Readable } from 'node:stream';
import { pipeline } from 'node:stream/promises';

async function* parseCsv(source) {
  let rest = '';
  const decoder = new TextDecoder();
  for await (const chunk of source) {
    rest += decoder.decode(chunk, { stream: true });
    const lines = rest.split(/\r?\n/);
    rest = lines.pop();
    for (const line of lines) if (line) yield line.split(',');
  }
  rest += decoder.decode();
  if (rest) yield rest.split(',');
}

const rows = [];
await pipeline(
  Readable.from([Buffer.from('a,b\n1,'), Buffer.from('2\n3,4')]),
  parseCsv,
  async (source) => { for await (const row of source) rows.push(row); },
);
console.log(rows);

[ [ 'a', 'b' ], [ '1', '2' ], [ '3', '4' ] ]

import { Writable, getDefaultHighWaterMark } from 'node:stream';

console.log(getDefaultHighWaterMark(false), getDefaultHighWaterMark(true));

const w = new Writable({ objectMode: true, write(chunk, enc, cb) { cb(); } });
try {
  w.write(null);
} catch (err) {
  console.log(err.code);
}

65536 16
ERR_STREAM_NULL_VALUES

import { Readable, Writable } from 'node:stream';
import { pipeline } from 'node:stream/promises';

// 1. A failing write destroys the whole pipeline, source included.
const source = Readable.from([1, 2, 3]);
const failing = new Writable({
  objectMode: true,
  write(n, _encoding, callback) {
    if (n === 2) callback(new Error(`bad value ${n}`));
    else callback();
  },
  destroy(err, callback) {
    console.log('destroy called, err =', err?.message);
    callback(err);
  },
});
try {
  await pipeline(source, failing);
} catch (err) {
  console.log('caught:', err.message);
}
console.log('source destroyed:', source.destroyed);

// 2. construct() and destroy() bracket a resource.
const resource = new Readable({
  construct(callback) { console.log('open'); setImmediate(callback); },
  read() { this.push('x'); this.push(null); },
  destroy(err, callback) { console.log('close resource'); callback(err); },
});
console.log(await resource.toArray());

// 3. Cancellation with an AbortSignal.
const ac = new AbortController();
const cancelled = new Readable({ read() {}, signal: ac.signal });
cancelled.on('error', (err) => console.log(err.name, err.code));
ac.abort();

destroy called, err = bad value 2
caught: bad value 2
source destroyed: true
open
close resource
[ <Buffer 78> ]
AbortError ABORT_ERR

import test from 'node:test';
import assert from 'node:assert/strict';
import { Readable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { CsvParser } from './csv-parser.js';

test('CsvParser handles rows split across chunks', async () => {
  const rows = [];
  await pipeline(
    Readable.from(['a,b\n1', ',2\n3,4'], { objectMode: false }),
    new CsvParser(),
    async (source) => { for await (const row of source) rows.push(row); },
  );
  assert.deepEqual(rows, [['a', 'b'], ['1', '2'], ['3', '4']]);
});

✔ CsvParser handles rows split across chunks
ℹ pass 1
ℹ fail 0

import { Readable } from 'node:stream';
import { from, pull, text } from 'node:stream/iter';

// Stateless transform: receives Uint8Array[] per batch, and null as the flush signal.
const upper = (chunks) => {
  if (chunks === null) return null;
  return chunks.map((chunk) => new TextEncoder().encode(
    new TextDecoder().decode(chunk).toUpperCase(),
  ));
};

console.log(await text(pull(from('hello bookshelf'), upper)));

// With the flag on, a classic Readable is a valid source.
console.log(await text(from(Readable.from(['a', 'b'], { objectMode: false }))));

(node:12345) ExperimentalWarning: stream/iter is an experimental feature and might change at any time
HELLO BOOKSHELF
ab