var Transform = require('readable-stream').Transform;
|
var inherits = require('inherits');
|
var cyclist = require('cyclist');
|
var util = require('util');
|
|
var ParallelTransform = function(maxParallel, opts, ontransform) {
|
if (!(this instanceof ParallelTransform)) return new ParallelTransform(maxParallel, opts, ontransform);
|
|
if (typeof maxParallel === 'function') {
|
ontransform = maxParallel;
|
opts = null;
|
maxParallel = 1;
|
}
|
if (typeof opts === 'function') {
|
ontransform = opts;
|
opts = null;
|
}
|
|
if (!opts) opts = {};
|
if (!opts.highWaterMark) opts.highWaterMark = Math.max(maxParallel, 16);
|
if (opts.objectMode !== false) opts.objectMode = true;
|
|
Transform.call(this, opts);
|
|
this._maxParallel = maxParallel;
|
this._ontransform = ontransform;
|
this._destroyed = false;
|
this._flushed = false;
|
this._ordered = opts.ordered !== false;
|
this._buffer = this._ordered ? cyclist(maxParallel) : [];
|
this._top = 0;
|
this._bottom = 0;
|
this._ondrain = null;
|
};
|
|
inherits(ParallelTransform, Transform);
|
|
ParallelTransform.prototype.destroy = function() {
|
if (this._destroyed) return;
|
this._destroyed = true;
|
this.emit('close');
|
};
|
|
ParallelTransform.prototype._transform = function(chunk, enc, callback) {
|
var self = this;
|
var pos = this._top++;
|
|
this._ontransform(chunk, function(err, data) {
|
if (self._destroyed) return;
|
if (err) {
|
self.emit('error', err);
|
self.push(null);
|
self.destroy();
|
return;
|
}
|
if (self._ordered) {
|
self._buffer.put(pos, (data === undefined || data === null) ? null : data);
|
}
|
else {
|
self._buffer.push(data);
|
}
|
self._drain();
|
});
|
|
if (this._top - this._bottom < this._maxParallel) return callback();
|
this._ondrain = callback;
|
};
|
|
ParallelTransform.prototype._flush = function(callback) {
|
this._flushed = true;
|
this._ondrain = callback;
|
this._drain();
|
};
|
|
ParallelTransform.prototype._drain = function() {
|
if (this._ordered) {
|
while (this._buffer.get(this._bottom) !== undefined) {
|
var data = this._buffer.del(this._bottom++);
|
if (data === null) continue;
|
this.push(data);
|
}
|
}
|
else {
|
while (this._buffer.length > 0) {
|
var data = this._buffer.pop();
|
this._bottom++;
|
if (data === null) continue;
|
this.push(data);
|
}
|
}
|
|
|
if (!this._drained() || !this._ondrain) return;
|
|
var ondrain = this._ondrain;
|
this._ondrain = null;
|
ondrain();
|
};
|
|
ParallelTransform.prototype._drained = function() {
|
var diff = this._top - this._bottom;
|
return this._flushed ? !diff : diff < this._maxParallel;
|
};
|
|
module.exports = ParallelTransform;
|
