Testing filtering and transforming streams in Nodejs and TypeScript

streamTest.ts

import { pipeline, Readable, Transform, Writable } from "stream";

const makeFilter = <A>(filterFn: (x: A) => boolean) => {
  return new Transform({
    objectMode: true,
    transform: (data: A, _, done) => {
      try {
        const willPass = filterFn(data);
        return done(null, willPass ? data : null);
      } catch (ex) {
        return done(ex, null);
      }
    },
  });
};

const makeTransform = <A, B>(transformFn: (x: A) => B) => {
  return new Transform({
    objectMode: true,
    transform: (data: A, _, done) => {
      try {
        // Log the data coming in -> prove that the undefined values don't get here form the filter
        console.log(`in makeTransform, data: ${data}`);
        return done(null, transformFn(data));
      } catch (ex) {
        return done(ex, null);
      }
    },
  });
};

const makeWritable = () => {
  return new Writable({
    objectMode: true,
    write: (chunk, _, next) => {
      console.log(chunk);
      next();
    },
  });
};

const someReadable = Readable.from([-2, -1, 0, 1, 2]);

pipeline(
  someReadable,
  makeFilter<number>((x) => x >= 0), // Filter out negative values
  makeTransform<number, number>((x) => x), // Identity transform just to log the values coming in
  makeWritable(), // Just some writable here
  (err) => {
    if (err) {
      console.error(err); // Log potential errors here
    }
  }
);

// Logs:
// in makeTransform, data: 0
// 0
// in makeTransform, data: 1
// 1
// in makeTransform, data: 2
// 2

// => The filtered values, aka `done(null, undefined/null)` are not
// passed to the following stream at all.