Handle Data Streams in Node.js

In Node.js, a stream is an abstract interface that inherits from the EventEmitter class used to read from or write to a source in a continuous fashion.

Unlike buffers, streams allow you to efficiently process data in chunks instead of loading it into memory all at once.

This is particularly useful for large amounts of data that would exceed the available memory space, such as large files, or for data that is consumed over time, such as network packets.

Beyond that, streams can be piped together — just like command-line interface programs on Unix-like operating systems — to create more complex data flows by connecting the output of one stream to the input of another.

In Node.js, there are 4 fundamental stream types:

• Readable: to read data from a file or any readable stream source.
• Writable: to write data to a file or any writable stream destination.
• Duplex: to both read and write.
• Transform: to transform the data as it is read and written.

Create a readable stream

To read data from a file or file descriptor in continuous fashion, chunk by chunk, you can create a readable stream using the fs.createReadStream() static method:

const fs = require('node:fs');

const stream = fs.createReadStream(path, {
  fd?,
  encoding?,
  highWaterMark?
});

Where:

• path is the path to the target file.
• fd is an optional integer used as a file descriptor. If specified, the path string is ignored.
• encoding is an optional string representing the character encoding. If specified, it returns a string, otherwise a Buffer. Defaults to null.
• highWaterMark is an optional integer representing the size of the internal buffer. Defaults to 64KB for buffers and 16KB for strings.

Paused mode vs flowing mode

When creating a readable stream, it operates in one of two modes: paused or flowing.

Paused mode

In paused mode, data is explicitly pulled from the stream by the application, which is suitable when you need precise control over the data flow.

It is mostly used for parsing large files row by row or waiting for user input before processing the next chunk of data.

This is the default mode when creating a readable stream.

Flowing mode

In flowing mode, data is automatically read from the stream and passed to the application through data events, which is suitable when you want to process data as soon as it becomes available.

It is mostly used for processing data in real time or streaming multimedia content.

Read data in paused mode

To read data in paused mode, you can attach a listener function to the 'readable' event and call the read() method of the stream instance.

stream.on('readable', () => {
  let data;

  while ((data = stream.read(size?)) !== null) {
    // process data
  }
});

Where size is an optional integer representing the number of bytes to read. Defaults to the size of the data contained in the internal buffer.

When the end of the stream is reached and all the data has been transmitted, it will trigger the 'end' event.

stream.on('end', () => {
  // process end of stream
});

Note: When reading a large file, it may happen that the read() method temporarily returns a null value, indicating that it has consumed all content in the internal buffer.

Unless the 'end' event is emitted, this means that there may be more data yet to be buffered, in which case a new 'readable' event is emitted once there's more data in the buffer.

Example

Let's consider this CSV file named leads.csv located in the current directory:

first_name,last_name,email_address,phone_number
John,Doe,johndoe@gmail.com,123-456-7890
Jane,Smith,jane.smith@hotmail.com,987-654-3210
Alice,Johnson,alicejnsn@outlook.com,555-123-4567
Bob,Brown,bobby03@gmail.com,555-987-6543

Let's consider this script, that reads the content of the leads.csv file and extracts the value of the email_address column line by line:

const fs = require('node:fs');

function processRow(row) {
  const columns = row && row.split(',');
  const email = columns[2];
  
  if (email) {
    console.log(email);
  }
}

function processCSVFile(filename) {
  const stream = fs.createReadStream(filename, {
    encoding: 'utf8',
    highWaterMark: 100
  });
  
  let buffer = '';

  stream.on('readable', () => {
    let chunk;
    let rows;

    while((chunk = stream.read()) !== null) {
      buffer += chunk;
      rows = buffer.split('\n');

      for (let i = 0 ; i < rows.length - 1 ; i++) {
        processRow(rows[i]);
      }

      buffer = rows.pop();
    }
  });

  stream.on('end', () => {
    processRow(buffer);
  });

  stream.on('error', error => {
    console.error(error);
  });
}

processCSVFile('leads.csv');

When executed, it will:

1. Create a readable stream from the file passed as argument to the processCSVFile() function.
2. Read the file chunk by chunk in paused mode.
3. Concatenate the newly transmitted chunk to the content of the buffer variable and split it into an array of strings using the newline character \n as delimiter.
4. Extract and output the email address of each string present in the rows array using the processRow() function, except for the last one, which may be incomplete.
5. Extract the last string from the rows array into the buffer variable and repeat the process until the stream emits an 'end' event.
6. Process the string present in the buffer variable using the processRow() function upon emission of the 'end' event.

Which will produce this output:

email_address
johndoe@gmail.com
jane.smith@hotmail.com
alicejnsn@outlook.com
bobby03@gmail.com

Read data in flowing mode

To read data in flowing mode, you can attach a listener function to the 'data' event and directly process the chunk as soon as available.

stream.on('data', chunk => {
  // process data chunk
});

When the end of the stream is reached, it will trigger the 'end' event.

stream.on('end', () => {
  // process end of stream
});

Create a writable stream

To write to a file or file descriptor in continuous fashion, you can create a writable stream using the fs.createWriteStream() static method:

const fs = require('node:fs');

const stream = fs.createWriteStream(path, {
  fd?,
  flags?,
  encoding?,
  highWaterMark?
});

Where:

• path is the path to the target file.
• fd is an optional integer representing a file descriptor. If specified, the path string is ignored.
• flags is an optional string representing the mode in which the file is opened. Defaults to w (i.e. truncate).
• encoding is an optional string representing the character encoding. Defaults to null.
• highWaterMark is an optional integer representing the size of the internal buffer. Defaults to 64KB (e.g., 64 characters in 'UTF-8').

Write data to a stream

To write data to a stream, you can use the write() method of the stream instance:

stream.write(chunk, encoding?, callback?);

Where:

• chunk is a string, Buffer, or Uint8Array.
• encoding is an optional string representing the character encoding. Defaults to utf8.
• callback is an optional function executed every time data is flushed.

To close the stream, you can use the end() method of the stream instance:

stream.end();

Example

Let's consider this script, that generates a random log file:

const fs = require('fs');

function generateLogEntry() {
  const levels = [
    '[ INFO ]',
    '[ WARNING ]',
    '[ ERROR ]'
  ];
  const messages = [
    'Application started successfully',
    'Connection to database established',
    'User logged in',
    'Invalid user input detected',
    'File not found',
    'Memory usage high',
    'Service unavailable',
    'Disk space running low',
    'Unexpected server shutdown',
    'Error fetching data from API'
  ];
  const level = levels[Math.floor(Math.random() * levels.length)];
  const message = messages[Math.floor(Math.random() * messages.length)];

  return `${level} ${message}\n`;
}

function generateLogFile(filePath, lines = 5) {
  const stream = fs.createWriteStream(filePath, {
    flags: 'a'
  });

  for (let i = 0; i < lines; i++) {
    stream.write(generateLogEntry());
  }

  stream.end();

  stream.on('error', error => {
    console.error(error.toString());
  });
}

generateLogFile('server.log');

When executed, it will:

1. Create a writable stream to the file passed as argument to the generateLogFile() function and open the file in append mode.
2. Generate random log lines using the generateLogEntry() function and write them to the stream.
3. Close the stream.

Which will produce this file:

$ cat server.log
[ INFO ] Memory usage high
[ INFO ] User logged in
[ INFO ] Application started successfully
[ INFO ] Connection to database established
[ INFO ] Disk space running low

Summary

Here's a summary of what you've learned in this lesson:

• A stream is an interface used to read from or write to a source in a continuous fashion, chunk by chunk.
• A readable stream is used to read data from a file or any readable stream source.
• A writable stream is used to write data to a file or any writable stream destination.
• The fs.createReadStream() static method is used to create a readable stream from a file or file descriptor.
• Readable streams operate in two modes: paused mode and flowing mode.
• In paused mode, the data is explicitly pulled from the readable stream by the application through the 'readable' event and the read() method of the stream instance.
• In flowing mode, the data is automatically sent to the readable stream and immediately consumed by the application through the 'data' event.
• The pause() and resume() methods of a readable stream instance are used to switch a readable stream from flowing mode to paused mode and vice versa.
• The 'end' event is used to signal the end of data transmission.
• The 'error' event is used to signal an error.
• The fs.createWriteStream() static method is used to create a writable stream to a file or file descriptor.
• The write() method of a writable stream instance is used to send data to the stream.
• The end() method of a writable stream instance is used to signal the end of data transmission.