stream-evolution.mjs

import { Readable, Transform, compose } from "stream";

{
  // Before
  const stream = new (class extends Readable {
    constructor() {
      super({ objectMode: true });
      this.data = [1, 2, 3];
    }
    _read() {
      this.push(this.data.shift() ?? null);
    }
  })();

  const mapped = stream.pipe(new class extends Transform {
      constructor() {
        super({ objectMode: true });
      }
      _transform(chunk, encoding, callback) {
        callback(null, chunk * 2);
      }
  }());
  // Real world usage should typically prefer listening to `readable` and 
  // calling read
  mapped.on("data", (data) => {
    console.log(data);
  });
}

{
  // Interim #1
  let data = [1, 2, 3];
  const stream = new Readable({
    objectMode: true,
    read() {
      this.push(data.shift() ?? null);
    },
  });
  const mapped = stream.pipe(new Transform({
    objectMode: true,
    transform(chunk, encoding, callback) {
      callback(null, chunk * 2);
    },
  }));
  mapped.on("data", (data) => {
    console.log(data);
  });
}

{
  // Interim #2 (modern)
  (async () => {
    const mapped = compose([1, 2, 3], async function* (input) {
      for await(const item of input) yield item * 2;
    });
    for await (const data of mapped) {
      console.log(data);
    }
  })();
}

{
  // After
  (async () => {
    for await (const data of Readable.from([1, 2, 3]).map((x) => x * 2)) {
      console.log(data);
    }
  })();
}