123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107 |
- var Stream = require('stream').Stream;
- var util = require('util');
- module.exports = DelayedStream;
- function DelayedStream() {
- this.source = null;
- this.dataSize = 0;
- this.maxDataSize = 1024 * 1024;
- this.pauseStream = true;
- this._maxDataSizeExceeded = false;
- this._released = false;
- this._bufferedEvents = [];
- }
- util.inherits(DelayedStream, Stream);
- DelayedStream.create = function(source, options) {
- var delayedStream = new this();
- options = options || {};
- for (var option in options) {
- delayedStream[option] = options[option];
- }
- delayedStream.source = source;
- var realEmit = source.emit;
- source.emit = function() {
- delayedStream._handleEmit(arguments);
- return realEmit.apply(source, arguments);
- };
- source.on('error', function() {});
- if (delayedStream.pauseStream) {
- source.pause();
- }
- return delayedStream;
- };
- Object.defineProperty(DelayedStream.prototype, 'readable', {
- configurable: true,
- enumerable: true,
- get: function() {
- return this.source.readable;
- }
- });
- DelayedStream.prototype.setEncoding = function() {
- return this.source.setEncoding.apply(this.source, arguments);
- };
- DelayedStream.prototype.resume = function() {
- if (!this._released) {
- this.release();
- }
- this.source.resume();
- };
- DelayedStream.prototype.pause = function() {
- this.source.pause();
- };
- DelayedStream.prototype.release = function() {
- this._released = true;
- this._bufferedEvents.forEach(function(args) {
- this.emit.apply(this, args);
- }.bind(this));
- this._bufferedEvents = [];
- };
- DelayedStream.prototype.pipe = function() {
- var r = Stream.prototype.pipe.apply(this, arguments);
- this.resume();
- return r;
- };
- DelayedStream.prototype._handleEmit = function(args) {
- if (this._released) {
- this.emit.apply(this, args);
- return;
- }
- if (args[0] === 'data') {
- this.dataSize += args[1].length;
- this._checkIfMaxDataSizeExceeded();
- }
- this._bufferedEvents.push(args);
- };
- DelayedStream.prototype._checkIfMaxDataSizeExceeded = function() {
- if (this._maxDataSizeExceeded) {
- return;
- }
- if (this.dataSize <= this.maxDataSize) {
- return;
- }
- this._maxDataSizeExceeded = true;
- var message =
- 'DelayedStream#maxDataSize of ' + this.maxDataSize + ' bytes exceeded.'
- this.emit('error', new Error(message));
- };
|