forked from enviPath/enviPy
99 lines
2.2 KiB
JavaScript
99 lines
2.2 KiB
JavaScript
'use strict';
|
|
|
|
var Duplex = require('readable-stream').Duplex;
|
|
var util = require('util');
|
|
|
|
// Inherit of Duplex stream
|
|
util.inherits(BufferStream, Duplex);
|
|
|
|
// Constructor
|
|
function BufferStream(options, cb) {
|
|
var _this = this;
|
|
|
|
// Ensure new were used
|
|
if (!(this instanceof BufferStream)) {
|
|
return new BufferStream(options, cb);
|
|
}
|
|
|
|
// Cast args
|
|
if(options instanceof Function) {
|
|
cb = options;
|
|
options = {};
|
|
}
|
|
options = options || {};
|
|
if(!(cb instanceof Function)) {
|
|
throw new Error('The given callback must be a function.');
|
|
}
|
|
this.__objectMode = options.objectMode;
|
|
|
|
// Parent constructor
|
|
Duplex.call(this, options);
|
|
|
|
// Keep a reference to the callback
|
|
this._cb = cb;
|
|
|
|
// Add a finished flag
|
|
this._bufferStreamFinished = false;
|
|
|
|
// Internal buffer
|
|
this._bufferStreamBuffer = [];
|
|
|
|
// Internal logic
|
|
function _bufferStreamCallbackWrapper(err) {
|
|
var buffer = options.objectMode ?
|
|
_this._bufferStreamBuffer :
|
|
Buffer.concat(_this._bufferStreamBuffer);
|
|
|
|
err = err || null;
|
|
_this._cb(
|
|
err,
|
|
buffer,
|
|
function(err2, buf) {
|
|
setImmediate(function() {
|
|
_this.removeListener('error', _bufferStreamError);
|
|
if(err2) {
|
|
_this.emit('error', err2);
|
|
}
|
|
_this._bufferStreamBuffer = options.objectMode ? buf || [] : [buf];
|
|
_this._bufferStreamFinished = true;
|
|
_this._read();
|
|
});
|
|
}
|
|
);
|
|
}
|
|
|
|
function _bufferStreamError(err) {
|
|
if(_this._bufferStreamFinished) {
|
|
return;
|
|
}
|
|
_bufferStreamCallbackWrapper(err);
|
|
}
|
|
|
|
this.once('finish', _bufferStreamCallbackWrapper);
|
|
|
|
this.on('error', _bufferStreamError);
|
|
}
|
|
|
|
BufferStream.prototype._write = function _bufferStreamWrite(chunk, encoding, done) {
|
|
this._bufferStreamBuffer.push(chunk);
|
|
done();
|
|
};
|
|
|
|
BufferStream.prototype._read = function _bufferStreamRead(n) {
|
|
var _this = this;
|
|
|
|
if(_this._bufferStreamFinished) {
|
|
while(_this._bufferStreamBuffer.length) {
|
|
if(!_this.push(_this._bufferStreamBuffer.shift())) {
|
|
break;
|
|
}
|
|
}
|
|
if(0 === _this._bufferStreamBuffer.length) {
|
|
_this.push(null);
|
|
}
|
|
}
|
|
|
|
};
|
|
|
|
module.exports = BufferStream;
|