How do I write a Node.js module to handle an incoming piped stream

Question

I'm trying to write a node module that accepts an incoming piped binary (or base-64-encoded) stream, but frankly I don't even know where to start. I can't see any examples in the Node docs about handling incoming streams; I only see examples on consuming them?

Say for example I want to be able to do this:

var asset = new ProjectAsset('myFile', __dirname + '/image.jpg')
var stream = fs.createReadStream(__dirname + '/image.jpg', { encoding: 'base64' }).pipe(asset)
stream.on('finish', function() {
    done()
})

I've gotten ProjectAsset looking like this, but I'm at a loss of where to go next:

'use strict'

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

var ProjectAsset = function() {
    var self = this

    Object.defineProperty(self, 'binaryData', {
        configurable: true,
        writable: true
    })

    stream.Stream.call(self)

    self.on('pipe', function(src) {
        // does it happen here? how do I set self.binaryData?
    })

    return self
}

util.inherits(ProjectAsset, stream.Stream)

module.exports = ProjectAsset
module.exports.DEFAULT_FILE_NAME = 'file'

this might help to mention that the buffer module can handle binary data so if youre reading and translating the binary in the same module you might want to look at the buffer module — Binvention
– Binvention, Commented Feb 2, 2016 at 17:03
I'm actually using Buffer to store the data when setting manually anyway - I need to translate the piped stream to a buffer so I can determine content type and accurate length. — brandonscript
– brandonscript, Commented Feb 2, 2016 at 17:06
so if you know how to use the stream api one of the options is the even listener on "data" if you just append that data to your buffer it should get what youre looking for — Binvention
– Binvention, Commented Feb 2, 2016 at 17:10

brandonscript · Accepted Answer · 2016-02-03 21:08:58Z

4

+50

It is possible to inherit from stream.Stream and make it work, however based on what's available in the documentation I would suggest inheriting from stream.Writable. Piping into a stream.Writable you'll need to have _write(chunk, encoding, done) defined to handle the piping. Here is an example:

var asset = new ProjectAsset('myFile', __dirname + '/image.jpg')
var stream = fs.createReadStream(__dirname + '/image.jpg', { encoding: 'base64' }).pipe(asset)
stream.on('finish', function() {
    console.log(asset.binaryData);
})

Project Asset

'use strict'

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

var ProjectAsset = function() {
    var self = this

    self.data
    self.binaryData = [];

    stream.Writable.call(self)

    self._write = function(chunk, encoding, done) {
        // Can handle this data however you want
        self.binaryData.push(chunk.toString())
        // Call after processing data
        done()
    }
    self.on('finish', function() {
        self.data = Buffer.concat(self.binaryData)
    })

    return self
}

util.inherits(ProjectAsset, stream.Writable)

module.exports = ProjectAsset
module.exports.DEFAULT_FILE_NAME = 'file'

If you're looking to also read from the stream, take a look at inheriting from stream.Duplex and also including the _read(size) method.

There's also the simplified constructors api if you're doing something simpler.

edited Feb 3, 2016 at 21:08

brandonscript

73.7k35 gold badges178 silver badges240 bronze badges

answered Jan 27, 2016 at 21:53

pohlman

3311 silver badge6 bronze badges

Sign up to request clarification or add additional context in comments.

5 Comments

brandonscript Over a year ago

Sorry it took so long, but so far this looks like the way to go. Only last bit is that the binaryData is still an array of bytes; somehow I need to convert that to a Buffer as well so that I can infer content type from it etc.

brandonscript Over a year ago

Hrm, and when I try fs.writeFile(__dirname + '/image2.jpg', asset.binaryData), the output file is not the same as the input. In fact, it's not even renderable as a jpg.

pohlman Over a year ago

For this you'll want to look at the Buffer class in node. I haven't messed with it much, but you can try starting by initializing the buffer as self.binaryData = new Buffer(''); From here you'll need to add data into the buffer doing something like this in your _write function: self.binaryData = Buffer.concat([self.binaryData, new Buffer(chunk, encoding)]). You may need to play with the encodings and try explicitly setting an encoding on the writeFile call.

brandonscript Over a year ago

So the missing piece here was that I needed to call Buffer.concat() on the binary data array. I'm going to have to figure out what to do about the bounty now, since both you and @Binvention helped equally!

brandonscript Over a year ago

Just added in the last piece that worked for me into your answer.

Binvention · Accepted Answer · 2016-02-02 17:33:18Z

2

Im not sure if this is exaclty what you were looking for but i think you could handle it using the buffer api with Buffer.concat on an array of buffers that can be retrieved form chunk on the stream data listener

'use strict'

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

var ProjectAsset = function() {
    var self = this

    Object.defineProperty(self, 'binaryData', {
        configurable: true,
        writable: true
    })

    stream.Stream.call(self)
    var data;
    var dataBuffer=[];
    self.on('data', function(chunk) {
        dataBuffer.push(chunk);
    }).on('end',function(){
        data=Buffer.concat(dataBuffer);
    });
    self.binaryData=data.toString('binary');
    return self
}

util.inherits(ProjectAsset, stream.Stream)

module.exports = ProjectAsset
module.exports.DEFAULT_FILE_NAME = 'file'

answered Feb 2, 2016 at 17:33

Binvention

1,0771 gold badge9 silver badges24 bronze badges

2 Comments

brandonscript Over a year ago

Nice! Not quite a working example, but in the end, I took @pohlman's code, and simply used Buffer.concat on the incoming buffer array in the end.

Binvention Over a year ago

Yeah sorry I didn't have the time to test the code but glad I could help

saintedlama · Accepted Answer · 2016-02-04 14:59:05Z

2

Since your using var asset = new ProjectAsset('myFile', __dirname + '/image.jpg') I suppose your ProjectAsset responsibility is to take some input stream do some transformations and write that to a file. You could implement a transform stream because you receive some input from a stream and generate some output of it that can be saved to a file or to some other write stream.

You could of course implement a transform stream by inheriting from node.js Transform Stream but inheriting is quite cumbersome so my implementation uses through2 to implement the transform stream:

module.exports = through2(function (chunk, enc, callback) {
  // This function is called whenever a piece of data from the incoming stream is read
  // Transform the chunk or buffer the chunk in case you need more data to transform

  // Emit a data package to the next stream in the pipe or omit this call if you need more data from the input stream to be read
  this.push(chunk);

  // Signal through2 that you processed the incoming data package
  callback();
 }))

Usage

var stream = fs.createReadStream(__dirname + '/image.jpg', { encoding: 'base64' })
               .pipe(projectAsset)
               .pipe(fs.createWriteStream(__dirname + '/image.jpg'));

As you can see in this example implementing a stream pipeline fully decouples data transformation and saving of the data.

Factory Function

If you prefer to use a constructor like approach in the project asset module because you need to pass some values or things you could easily export a constructor function as shown below

var through2 = require('through2');

module.exports = function(someData) {

  // New stream is returned that can use someData argument for doing things
  return through2(function (chunk, enc, callback) {
    // This function is called whenever a piece of data from the incoming stream is read
    // Transform the chunk or buffer the chunk in case you need more data to transform

    // Emit a data package to the next stream in the pipe or omit this call if you need more data from the input stream to be read
    this.push(chunk);

    // Signal through2 that you processed the incoming data package
    callback();
  });
}

Usage

var stream = fs.createReadStream(__dirname + '/image.jpg', { encoding: 'base64' })
               .pipe(projectAsset({ foo: 'bar' }))
               .pipe(fs.createWriteStream(__dirname + '/image.jpg'));

edited Feb 4, 2016 at 14:59

answered Feb 1, 2016 at 12:43

saintedlama

6,8781 gold badge30 silver badges47 bronze badges

6 Comments

brandonscript Over a year ago

I'm thinking of this more as a proxy - less likely to load from fs and more likely to load from request. I'm using fs though just to construct the function and write tests.

brandonscript Over a year ago

I like the idea of this, but in the interests of keeping 3rd party dependencies to a min., I'm going to implement it natively. Thanks for the tip on through2 though!

edin-m Over a year ago

@remus through2 is better, less likely to mess things up. And easier. YMMW though.

brandonscript Over a year ago

Aright, well first, I can't wrap my class in through2 - I return the class not an instance of it in module.exports do that I can do var asset = new ProjectAsset

saintedlama Over a year ago

You could return a factory function from module.exports instead of a constructor which encapsulates how an instance is created. My tip: Don't OOP to much in JavaScript :)

|

Collectives™ on Stack Overflow

How do I write a Node.js module to handle an incoming piped stream

3 Answers 3

5 Comments

2 Comments

6 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

5 Comments

2 Comments

6 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related