What are Streams in Node.js?
Streams allows for more efficient processing of large amounts of data, and can be viewed as a “river” of data. Streams “saves” memory because by processing data in chunks, working with large amounts of data does not occupy a large amount of memory at once. Streams processes data quickly, because the data can be processed as soon as the first piece arrives, without waiting for the complete data set, which speeds up the processing process.
Types of Streams
- Readable Streams: Enable data to be read, like water flowing from a faucet.
- Writable Streams: Enable data to be written, like pouring water into a sink.
- Duplex Streams: Can simultaneously read and write data, similar to using a telephone.
- Transform Streams: A special type of Duplex Streams that can transform data as it passes through the stream, similar to a water filter.
Streams in Node.js are a powerful tool for efficient real-time data processing. Their ability to process large amounts of data in chunks, while conserving resources and speeding up processes, makes them indispensable in application development.
Writing
Writing with the help of streams is best explained with examples, and we will write programs (with and without the use of streams) that aim to write a million numbers, each in a new line, into the corresponding file.
NOTE: Code execution using streams is up to 30 times faster!!!
Example without using streams
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
const fs = require('fs').promises; (async () => { try { const fd = await fs.open('bezStreamsPromisesBuffer.txt', 'w'); for (let i = 0; i < 1000000; i++) { // We create a buffer with the data we want to write const buffer = Buffer.from(`Broj: ${i}n`, 'utf8'); // We write the buffer to the file asynchronously await fs.write(fd, buffer); } await fd.close(); console.log('File saved successfully using promises and Buffer.'); } catch (error) { console.error('An error occurred:', error); } })(); |
Example using streams
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
const fs = require('fs'); const stream = fs.createWriteStream('saStreams.txt'); for (let i = 0; i < 1000000; i++) { stream.write(`Broj: ${i}n`); } stream.end(); stream.on('finish', () => { console.log('File successfully saved with Streams!'); }); stream.on('error', (err) => { console.error('An error occurred:', err); }); |
In the previous example, we don’t use a buffer, so the string “Number: ${i}n” is implicitly converted to binary numbers before being written to the stream, but this can be written differently (“more efficiently”), if a Buffer object is explicitly created using the Buffer.from() method and then used to convert strings to Buffer objects before being written to the stream:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
const fs = require('fs'); const stream = fs.createWriteStream('saStreams.txt'); for (let i = 0; i < 1000000; i++) { // Explicit buffer creation const buffer = Buffer.from(`Broj: ${i}n`, 'utf8'); stream.write(buffer); } stream.end(); stream.on('finish', () => { console.log('File successfully saved with Streams!'); }); stream.on('error', (err) => { console.error('An error occurred:', err); }); |
Drain event
If you are using a writable stream to write data to an output, you may be writing faster than the output can accept the data, and this may cause data to accumulate in the memory buffer. To avoid this, the procedure is as follows: when writing data to a writable stream, we can monitor whether the stream has emptied its buffer using the “drain” event. When the “drain” event is emitted, it means that the stream is signaling that it is ready to accept more data, and only then can we continue writing data to the stream without fear of buffer overload.
Example
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
const stream = someWritableStream(); function writeData(data) { if (!stream.write(data)) { console.log('Buffer full, waiting for drain event...'); stream.once('drain', () => { console.log('The buffer is empty, you can continue writing.'); writeData(noviPodaci); // We continue writing after the buffer is empty }); } } // Usage examples writeData(nekiPodaci); |
In this example, when stream.write(data) returns false, it means that the stream buffer is full and we should wait for a “drain” event before continuing to write. When the “drain” event is emitted, we can safely continue writing data.