Readable, Writable and Transformable Streams

The more I’ve written Node.js professionally and as a hobbyist, the more I’ve been of the opinion that streams and event emitters are really special. They are not special in the sense they they are unique to the framework; they are special because Node.js' API’s for them make them useful. Streams are, in my opinion, the most useful data structure in Node. With that being said, I wanted to share some notes on streams. I built up a trivial example using readable, writable and transformable streams in Node.js.

Readable Stream

The readable stream is a stream that reads in data (duh). Think of it as a gas tank (hang with me here). You’ll pour gasoline into the tank from a source like a pump at the gas station. The tank is always read to receive gas (read). The weakness of this analogy is that the tank can become full and streams don’t have that property. We can get the basic idea by creating a readable stream with the Node.js class for them.

var stream = require('stream');

class ReadableStream extends stream.Readable { constructor(){ super(); stream.Readable.call(this, {objectMode: true}); // Janky as hell JS }

_read(value){ console.log("Reading data in"); } }

var myReadStream = new ReadableStream();

myReadStream.on('error', (data)=>{ console.log("There was an error with the data", data) });

myReadStream.on('data', (data)=>{ console.log(data) });

myReadStream.on('end', ()=>{ console.log("no more data") });

myReadStream.push("Hello"); myReadStream.push(null);

Writable Stream

The writable stream uses the data that is given to it. Sticking with our car analogy, its where is the gas in the gas tank going? It’s going to the combustion system. Here’s an example:

var stream = require('stream');

class WritableStream extends stream.Writable { constructor(){ super(); stream.Writable.call(this, {objectMode: true}); // Janky as hell JS }

_write(value, encoding, callback){ console.log("Writing a value..."); callback(); } }

var myWriteStream = new WritableStream();

myWriteStream.on('write', (data)=>{ console.log('wrote some data ' +data) })

myWriteStream.on('error', (error)=>{ console.log('Error writing data ' +error) })

Transformable Stream

This is a unique stream in that it is both a readable and writable stream. Meaning it can be read into and written out of. Keeping with our terrible gas tank analogy, it’s like the engine. Fuel goes in and is turned into heat and used to power the car. Here’s an example.

var stream = require('stream');

// Function to reverse a string function reverse(s) { return s.split('').reverse().join(''); }

class TransformableStream extends stream.Transform { constructor(){ super(); stream.Transform.call(this, {objectMode: true}); // Janky as hell JS }

_transform(value, encoding, callback){ this.push(reverse(value)) callback(); } }

All Working Together:

var stream = require('stream');

class ReadableStream extends stream.Readable { constructor(){ super(); stream.Readable.call(this, {objectMode: true}); // Janky as hell JS }

_read(value, encoding){ console.log("Reading data in..."); } }

var myReadStream = new ReadableStream();

myReadStream.on('error', (data)=>{ console.log("There was an error with the data", data) });

myReadStream.on('data', ()=>{ });

myReadStream.on('end', ()=>{ console.log("no more data") });

class WritableStream extends stream.Writable { constructor(){ super(); stream.Writable.call(this, {objectMode: true}); }

_write(value){ console.log("Writing a value: "+ value); } }

var myWriteStream = new WritableStream();

myWriteStream.on('write', (data)=>{ console.log('wrote some data: '+data) })

myWriteStream.on('error', (error)=>{ console.log('Error writing data', error) })

function reverse(s) { return s.split('').reverse().join(''); }

class TransformableStream extends stream.Transform { constructor(){ super(); stream.Transform.call(this, {objectMode: true}); }

_transform(value, encoding, callback){ this.push(reverse(value)) callback(); } }

var myTransformStream = new TransformableStream();

myReadStream.pipe(myTransformStream).pipe(myWriteStream);

myReadStream.push("Goodbye") myReadStream.push(null);

Notes

I used the ES2015 syntax for classes versus the traditional Javascript way of using functions. In practice it’s just syntactical sugar. There’s nothing really special about this way except that it’s more concise. Many smart people argue against this way of doing programming in Javascript, that you should use Prototypal inheritance only. I’m not really on that train of thought. Also, I know these examples are contrived but you can easily adjust your read method to handle a file, a web socket or a stream of JSON. What I’ve found is the simplest examples often help comprehension the most.

[1] If you have trouble running the examples, the full notebook is here.