How to implement a writable stream
Asked Answered
P

3

80

I want to pipe data from an amazon kinesis stream to a an s3 log or a bunyan log.

The sample works with a file write stream or stdout. How would I implmeny my own writable stream?

//this works
var file = fs.createWriteStream('my.log')
kinesisSource.pipe(file)

this doesn't work saying it has no method 'on'

var stream = {}; //process.stdout works however
stream.writable = true;
stream.write =function(data){
    console.log(data);
};
kinesisSource.pipe(stream);

what methods do I have to implement for my own custom writable stream, the docs seem to indicate I need to implement 'write' and not 'on'

Puberulent answered 31/1, 2014 at 22:55 Comment(0)
P
166

To create your own writable stream, you have three possibilities.

Create your own class

For this you'll need:

  1. To extend the Writable class.
  2. To call the Writable constructor in your own constructor.
  3. To define a _write() method in the prototype of your stream object.

Here's an example:

var stream = require('stream');
var util = require('util');

function EchoStream () { // step 2
  stream.Writable.call(this);
};
util.inherits(EchoStream, stream.Writable); // step 1
EchoStream.prototype._write = function (chunk, encoding, done) { // step 3
  console.log(chunk.toString());
  done();
}

var myStream = new EchoStream(); // instanciate your brand new stream
process.stdin.pipe(myStream);

Extend an empty Writable object

Instead of defining a new object type, you can instanciate an empty Writable object and implement the _write() method:

var stream = require('stream');
var echoStream = new stream.Writable();
echoStream._write = function (chunk, encoding, done) {
  console.log(chunk.toString());
  done();
};

process.stdin.pipe(echoStream);

Use the Simplified Constructor API

If you're using io.js, you can use the simplified constructor API:

var writable = new stream.Writable({
  write: function(chunk, encoding, next) {
    console.log(chunk.toString());
    next();
  }
});

Use an ES6 class in Node 4+

class EchoStream extends stream.Writable {
  _write(chunk, enc, next) {
    console.log(chunk.toString());
    next();
  }
}
Phospholipide answered 5/2, 2014 at 17:9 Comment(4)
to support object mode replace chunk.toString() per chunk.toString ? chunk.toString() : chunkFeminism
What does this line do: "util.inherits(EchoStream, stream.Writable); // step 1 "?Cableway
It makes EchoStream a "sub-class" of stream.Writable: its prototype methods are inherited from it and stream.Writable is accessible using the super_ property. See the documentation for more information.Phospholipide
What's the benefit of "Create your own class" over "Extend an empty Writable object"?Conchoidal
S
11

Actually to create a writeable stream is quite simple. Here's is the example:

var fs = require('fs');
var Stream = require('stream');

var ws = new Stream;
ws.writable = true;
ws.bytes = 0;

ws.write = function(buf) {
   ws.bytes += buf.length;
}

ws.end = function(buf) {
   if(arguments.length) ws.write(buf);
   ws.writable = false;

   console.log('bytes length: ' + ws.bytes);
}

fs.createReadStream('file path').pipe(ws);

Also if you want to create your own class, @Paul give a good answer.

Schlieren answered 18/2, 2014 at 2:26 Comment(4)
Cannot read property 'length' of undefined... at the buf.length in ws.write function definitionPhina
should be buf.lengthZeitler
This may have been correct at one time, but not anymore. The Node documentation now says: "The stream.Writable class is extended to implement a Writable stream. Custom Writable streams must call the new stream.Writable([options]) constructor and implement the writable._write() and/or writable._writev() method." nodejs.org/api/…Afflict
This is completely wrong both logically and syntacticallyRiti
C
5

Here is an example directly from nodejs docs
https://nodejs.org/api/stream.html#an-example-writable-stream

const { Writable } = require('stream');
class MyWritable extends Writable {
  _write(chunk, encoding, callback) {
    if (chunk.toString().indexOf('a') >= 0) {
      callback(new Error('chunk is invalid'));
    } else {
      callback();
    }
  }
}
Chameleon answered 28/1, 2022 at 21:43 Comment(0)

© 2022 - 2024 — McMap. All rights reserved.