Node.js stream API

The Node.js stream API is scalable and offers the greatest control over the data flow.

Using the pipe API

Pipes in Node.js is a native functionnality provided by the stream API. It behave just like Unix pipes where the output of a process, here a stream reader, is redirected as the input of the following process, here a stream writer.

The pipe example is quite readable while also scalable:

// Import the package
import * as csv from 'csv';

// Run the pipeline
csv
  // Generate 20 records
  .generate({
    delimiter: '|',
    length: 20
  })
  // Transform CSV data into records
  .pipe(csv.parse({
    delimiter: '|'
  }))
  // Transform each value into uppercase
  .pipe(csv.transform((record) =>
    record.map((value) =>
      value.toUpperCase()
    )
  ))
  // Convert objects into a stream
  .pipe(csv.stringify({
    quoted: true
  }))
  // Print the CSV stream to stdout
  .pipe(process.stdout);

./packages/csv/samples/pipe.js

Using the native stream functions

The native stream functions provide flexibity but comes at the cost of being more verbose and harder to write. Data is consumed inside the readable event with the stream.read function. It is then written by calling the stream.write function. The stream example illustrates how to initialize each packages and how to plug them.

// Import the package
import {generate, parse, transform, stringify} from 'csv';

let i = 0;

const generator = generate({seed: 1, columns: 2, length: 20});
const parser = parse();
const transformer = transform(function(data){
  i++;
  return data.map(function(value){return value.toUpperCase();});
});
const stringifier = stringify();

// Read generated CSV data and send it to the parser
generator.on('readable', function(){
  let data; while((data = generator.read()) !== null){
    parser.write(data);
  }
});
// When generation is over, close the parser
generator.on('end', function(){
  parser.end();
});

// Read parsed records and send them to the transformer
parser.on('readable', function(){
  let data; while((data = parser.read()) !== null){
    transformer.write(data);
  }
});
// When parsing is over, close the transformer
parser.on('end', function(){
  transformer.end();
});

// Read transformed records and send them to the stringifier
transformer.on('readable', function(){
  let data; while((data = transformer.read()) !== null){
    stringifier.write(data);
  }
});
// When transformation is over, close the stringifier
transformer.on('end', function(){
  stringifier.end();
});

// Read CSV data and print it to stdout
stringifier.on('readable', function(){
  let data; while((data = stringifier.read()) !== null){
    process.stdout.write(data);
  }
});
// When stringifying is over, print a summary to stderr
generator.on('close', function(){
  process.stderr.write('=> ' + i + ' records\n');
});

./packages/csv/samples/pipe.js