throughv
stream.Transform with parallel chunk processingSame as Rod Vagg's through2 but with parallel chunk processing.
throughv is useful to
augment/process data
coming from a stream in a fast parallel fashion, e.g. fetching some
relevant data from a database.fs.createReadStream('ex.txt')
.pipe(throughv(function (chunk, enc, callback) {
// this happen in parallel for all chunks
// in the stream's buffer, the parallelism
// is determined by highWaterMark
for (var i = 0; i < chunk.length; i++)
if (chunk[i] == 97)
chunk[i] = 122 // swap 'a' for 'z'
setImmediate(callback, null, chunk)
}))
.pipe(fs.createWriteStream('out.txt'))Or object streams:
var all = []
fs.createReadStream('data.csv')
.pipe(csv2())
.pipe(throughv.obj(function (chunk, enc, callback) {
// this happen in parallel for all chunks
// in the stream's buffer, the parallelism
// is determined by highWaterMark
var data = {
name : chunk[0]
, address : chunk[3]
, phone : chunk[10]
}
setImmediate(callback, null, data)
}))
.on('data', function (data) {
all.push(data)
})
.on('end', function () {
doSomethingSpecial(all)
})Note that
throughv.obj(fn) is a convenience wrapper around `throughv({
objectMode: true }, fn)`.Install
npm i throughv --saveAPI
throughv( options, transformFunction , flushFunction
)Consult the stream.Transform documentation for the exact rules of the
transformFunction (i.e.
this._transform) and the optional flushFunction (i.e.
this._flush).options
The options argument is optional and is passed straight through tostream.Transform. So you can use objectMode:true if you are
processing non-binary streams (or just use throughv.obj()).In order to set the maximum parallelism at which the instance will process chunks, set highWaterMark. It is defaulted at 16KB for binary streams, and at 16 for object streams.
The
options argument is first, unlike standard convention, because if
I'm passing in an anonymous function then I'd prefer for the options
argument to not get lost at the end of the call:fs.createReadStream('/tmp/important.dat')
.pipe(throughv({ objectMode: true, allowHalfOpen: false },
function (chunk, enc, cb) {
cb(null, 'wut?')
}
)
.pipe(fs.createWriteStream('/tmp/wut.txt'))transformFunction
ThetransformFunction must have the following signature: `function
(chunk, encoding, callback) {}`. A minimal implementation should call
the callback function to indicate that the transformation is done,
even if that transformation means discarding the chunk.To queue a new chunk, call
callback(err, data) —this must be called
only once for each chunk.If you do not provide a
transformFunction then you will get a
simple pass-through stream.flushFunction
The optionalflushFunction is provided as the last argument (2nd or
3rd, depending on whether you've supplied options) is called just prior
to the stream ending. Can be used to finish up any processing that may
be in progress.fs.createReadStream('/tmp/important.dat')
.pipe(throughv(
function (chunk, enc, cb) { cb(null, chunk) }, // transform is a
noop
function (cb) { // flush function
this.push('tacking on an extra buffer to the end');
cb();
}
))
.pipe(fs.createWriteStream('/tmp/wut.txt'));new Throughv(options)This has the same api of Transform, so you can subclass it if you want
Acknowledgements
throughv is sponsored by nearForm.Code was taken and adapted from node.js, readable-stream, and through2.
