Node.js - Streams
Tags: node, javascript, eventdriven
Published: 2020-12-10 12:51:51

![Alt Text](https://dev-to-uploads.s3.amazonaws.com/i/zgxmg133amn90ftdfdo6.jpeg)

#### Overview
This article is **Part 4** of [Working With NodeJS](https://dev.to/meddy672/series/20844), a series of articles to help simplify learning Node.js. In this article I will cover **Streams**.


#### Introduction
Working with large amounts of data usually means working with streams. Streams allow for large data processing without requiring excessive compute resources. With streams you read data piece by piece, processing its content without keeping it all in memory.

#### Stream Basics
A stream is a collection of data that may not be available all at once. Think about watching a Youtube or Netflix video. When the video starts, it starts with just enough data to get the video going and over time the remaining data is processed. This type of data handling provides two major advantages over other data handling methods:
1. **Memory efficiency**: you don't need to load large amounts of data in memory before you are able to process it.
2. **Time efficiency**: it takes way less time to start processing data as soon as you have it, rather than waiting until the whole data payload is available to start.

With efficient data handling, we can deliver high volume data processing applications in Node.js. The [stream module](https://nodejs.org/dist/latest-v15.x/docs/api/stream.html#stream_stream) provides the foundation upon which all streaming APIs are built in Node.js. Within the stream module are four types of streams:  *Readable, Writable, Duplex*, and *Transform*. Each of these implement the `pipe` method, which takes the source of one stream and connects its output to another stream. The connecting of two or more streams can thought of as a `pipeline` where the data of a stream passes through each phase of the pipeline until reaching its final destination.

#### Readable Stream
A [readable](https://nodejs.org/dist/latest-v15.x/docs/api/stream.html#stream_readable_streams) stream could be used to read a file, read data from an incoming HTTP request, or read user input from a command prompt to name a few examples. The Readable constructor inherits from the Stream constructor which inherits from the EventEmitter constructor, so readable streams are event emitters. To create a readable stream, the Readable constructor is called with the `new` keyword and passed an options object with a read method.
##### code snippet
```javascript
'use strict'
const { Readable } = require('stream')
const createReadStream = () => {
  const data = ['some', 'data', 'to', 'read']
  return new Readable({
    encoding: 'utf8',
    read () {
      if (data.length === 0) this.push(null)
      else this.push(data.shift())
    }
  })
}
const readable = createReadStream()
readable.on('data', (data) => { console.log('got data', data) })
readable.on('end', () => { console.log('finished reading') })
```
##### output
```
got data some
got data data
got data to
got data read
finished reading
```

#### Writable Stream
 A [writable](https://nodejs.org/dist/latest-v15.x/docs/api/stream.html#stream_writable_streams) stream could be used to write a file, write data to an HTTP response, or write to the terminal. The Writable constructor inherits from the Stream constructor which inherits from the EventEmitter constructor, so writable streams are event emitters. To create a writable stream, call the Writable constructor with the `new` keyword. The options object of the Writable constructor can have a write function, which takes three arguments `chunk, enc, next`. To send data to a writable stream use the `write` method. The `end` method can be used write a final payload to the stream before ending it. Once the stream is ended the `finish` event is emitted.
```javascript
'use strict'
const { Writable } = require('stream')
const createWriteStream = (data) => {
  return new Writable({
    decodeStrings: false,
    write (chunk, enc, next) {
      data.push(chunk);
      next();
    }
  })
}
const data = [];
const writable = createWriteStream(data);
writable.on('finish', () => {console.log('finished', data)});
writable.write('A');
writable.write('B');
writable.write('C');
writable.end('nothing more to write');
```
##### output
```
finished [ 'A', 'B', 'C', 'nothing more to write' ]
```

#### Duplex Stream
[Duplex](https://nodejs.org/dist/latest-v15.x/docs/api/stream.html#stream_duplex_and_transform_streams) streams are streams that implement both the Readable and Writable interfaces. A good example of a Duplex stream would be a TCP socket. TCP socket can read data from a client connection as well as write back to the client. To demonstrate an example I'll create two files to simulate a TCP socket and client connection.
##### TCP server
```javascript
'use strict'
const net = require('net')
net.createServer((socket) => {
  const interval = setInterval(() => {
    socket.write('beat')
  }, 1000)
  socket.on('data', (data) => {
    socket.write(data.toString().toUpperCase())
  })
  socket.on('end', () => { clearInterval(interval) })
}).listen(3000)
```

##### client connection
```javascript
'use strict'
const net = require('net')
const socket = net.connect(3000)

socket.on('data', (data) => {
  console.log('got data:', data.toString())
})

setTimeout(() => {
  socket.write('all done')
  setTimeout(() => {
    socket.end()
  }, 250)
}, 3250)
```
Running both scripts will produce the following output:
##### output
```
got data: beat
got data: beat
got data: beat
got data: ALL DONE
```
#### Transform Stream
[Transform](https://nodejs.org/dist/latest-v15.x/docs/api/stream.html#stream_duplex_and_transform_streams) streams are duplex streams with an additional constraint applied to enforce the relationship between the read and write interfaces. The constraint between the read and write interfaces is enforced through the `transform` function. The transform function has the same signature as the write function from the Writable stream object in that it takes `chunk`,`enc`, and `next` as parameters. The difference is the `next` function can be passed a second argument which should be the result of applying some kind of transform operation to the incoming chunk. Lets see a quick example.
##### code snippet
```javascript
'use strict'
const { Transform } = require('stream')
const createTransformStream = () => {
  return new Transform({
    decodeStrings: false,
    encoding: 'utf8',
    transform (chunk, enc, next) {
     next(null, chunk.toUpperCase());
    }
  })
}
const transform = createTransformStream()
transform.on('data', (data) => {
  console.log('got data:', data);
})
transform.write('a\n');
transform.write('b\n');
transform.write('c\n');
transform.end('nothing more to write');
```
##### output
```
got data: A

got data: B

got data: C

got data: NOTHING MORE TO WRITE
```

#### Piping Streams
As stated before, the [pipe](https://nodejs.org/dist/latest-v15.x/docs/api/stream.html#stream_readable_pipe_destination_options) method takes the source of one stream and pipes it into the destination of another stream. Lets see a quick example. I'll refactor the *Readable* and *Writable* examples from the previous sections so they work together using the `pipe` method.
```javascript
'use strict'
const { Readable, Writable } = require('stream')
const createReadStream = () => {
  const readData = ['some', 'data', 'to', 'read'];
  return new Readable({
    encoding: 'utf8',
    read () {
      if (readData.length === 0) this.push(null)
      else this.push(readData.shift())
    }
  })
}

const createWriteStream = (data) => {
  return new Writable({
    decodeStrings: false,
    write (chunk, enc, next) {
      data.push(chunk);
      next();
    }
  })
}
const data = [];
const readable = createReadStream();
const writable = createWriteStream(data);
readable.pipe(writable);
writable.on('finish', () => {console.log('finished', data)});
```

##### output
```
finished [ 'some', 'data', 'to', 'read' ]
```
In the code snippet above the `readable.on` method has been removed from the code. This is important because streams are in a paused state by default. The only way to get data flowing is to use the *resume method*, *data event* or the *pipe method*. You can pipe as many streams as you need in order to satisfy your use case, however, it is best practice to use a pipeline if piping more then two streams.

A [pipeline](https://nodejs.org/dist/latest-v15.x/docs/api/stream.html#stream_stream_pipeline_streams_callback) can be used to pipe a series of streams together. Let see an example. I'll refactor the code from the *Readable, Writable, and Transform* sections so they work using a pipeline.
##### pipeline snippet
```javascript
'use strict'
const { Readable, Writable, Transform, pipeline } = require('stream')
const createReadStream = () => {
  const readData = ['some', 'data', 'to', 'read'];
  return new Readable({
    encoding: 'utf8',
    read() {
      if (readData.length === 0) this.push(null);
      else this.push(readData.shift());
    }
  })
}

const createTransform = () => {
  return new Transform({
    transform(chunk, enc, next) {
      const changedData = chunk.toString().toUpperCase();
      next(null, changedData);
    }
  })
}

const createWriteStream = () => {
  const data = [];
  const writable = new Writable({
    decodeStrings: false,
    write(chunk, enc, next) {
      data.push(chunk.toString());
      next();
    }
  });
  writable.data = data;
  return writable;
}

const readable = createReadStream();
const writable = createWriteStream();
const transform = createTransform();
pipeline(readable, transform, writable, (err) => {
  if (err) console.error('Pipeline failed.', err);
  else console.log('Pipeline succeeded.', writable.data);
});
```
##### output
```
Pipeline succeeded. [ 'SOME', 'DATA', 'TO', 'READ' ]
```
In the above code snippet, I imported the `pipeline` function from the stream module. Next, I used three variables, which reference streaming functions, to interact with the data flowing through the pipeline. Finally, the pipeline takes a callback with an `err` parameter that will execute once the pipeline completes. If an error occurs the pipeline will fail, otherwise the console will log the data with a success message.

There are much more to streams then the examples in this article. I always suggest using the [NodeJS Docs](https://nodejs.org/dist/latest-v15.x/docs/api/) as your primary source when developing a solution for your use case. There is also a lot of good articles written by other developers here on **DEV** that can help you when developing with streams in Node.js. Another good author is [Samer Buna](https://www.freecodecamp.org/news/author/samerbuna/). Samer has a lot good content on advanced topics about Node.js. As always if you have any questions post it in the discussions and I will reply. Take care and happy coding.