Piping Streams Together

The output of one readable stream can get connected, "piped", to the input of another writable stream or to a duplex stream.
As an example here:

process.stdin is a readable stream
this gets piped to a TransformStream to change the stdin contents to an array
this gets piped, again, to convert the array back to a string
this gets piped, again, to process.stdout (the terminal output)

const { Transform } = require('stream');

const commaSplitter = new Transform({
  readableObjectMode: true,

  transform(chunk, encoding, callback) {
    this.push(chunk.toString().trim().split(','));
    callback();
  },
});

const arrayToObject = new Transform({
  readableObjectMode: true,
  writableObjectMode: true,
  
  transform(chunk, encoding, callback) {
    const obj = {};
    for(let i=0; i < chunk.length; i+=2) {
      obj[chunk[i]] = chunk[i+1];
    }
    this.push(obj);
    callback();
  }
});

const toString = new Transform({
  writableObjectMode: true,
  transform(chunk, encoding, callback) {
    this.push(JSON.stringify(chunk) + '\n');
    callback();
  },
});

process.stdin
  .pipe(commaSplitter)
  .pipe(toString)
  .pipe(process.stdout);

Tags:

node

streams

core