npm install throughvstream.Transform with parallel chunk processing
Same 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.
``js
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:
`js
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).
npm i throughv --save
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).
The options argument is optional and is passed straight through to
stream.Transform. So you can use objectMode:true if you arethroughv.obj()
processing non-binary streams (or just use ).
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:
`js`
fs.createReadStream('/tmp/important.dat')
.pipe(throughv({ objectMode: true, allowHalfOpen: false },
function (chunk, enc, cb) {
cb(null, 'wut?')
}
)
.pipe(fs.createWriteStream('/tmp/wut.txt'))
The transformFunction must have the following signature: function
(chunk, encoding, callback) {}. A minimal implementation should callcallback
the 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.
The optional flushFunction 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.
`js``
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
throughv is sponsored by nearForm.
Code was taken and adapted from
node.js, readable-stream, and through2.
MIT