1 var util = require('util');
2 var Stream = require('stream').Stream;
3 var DelayedStream = require('delayed-stream');
5 module.exports = CombinedStream;
6 function CombinedStream() {
10 this.maxDataSize = 2 * 1024 * 1024;
11 this.pauseStreams = true;
13 this._released = false;
15 this._currentStream = null;
17 util.inherits(CombinedStream, Stream);
19 CombinedStream.create = function(options) {
20 var combinedStream = new this();
22 options = options || {};
23 for (var option in options) {
24 combinedStream[option] = options[option];
27 return combinedStream;
30 CombinedStream.isStreamLike = function(stream) {
31 return (typeof stream !== 'function')
32 && (typeof stream !== 'string')
33 && (typeof stream !== 'boolean')
34 && (typeof stream !== 'number')
35 && (!Buffer.isBuffer(stream));
38 CombinedStream.prototype.append = function(stream) {
39 var isStreamLike = CombinedStream.isStreamLike(stream);
42 if (!(stream instanceof DelayedStream)) {
43 var newStream = DelayedStream.create(stream, {
44 maxDataSize: Infinity,
45 pauseStream: this.pauseStreams,
47 stream.on('data', this._checkDataSize.bind(this));
51 this._handleErrors(stream);
53 if (this.pauseStreams) {
58 this._streams.push(stream);
62 CombinedStream.prototype.pipe = function(dest, options) {
63 Stream.prototype.pipe.call(this, dest, options);
68 CombinedStream.prototype._getNext = function() {
69 this._currentStream = null;
70 var stream = this._streams.shift();
73 if (typeof stream == 'undefined') {
78 if (typeof stream !== 'function') {
79 this._pipeNext(stream);
83 var getStream = stream;
84 getStream(function(stream) {
85 var isStreamLike = CombinedStream.isStreamLike(stream);
87 stream.on('data', this._checkDataSize.bind(this));
88 this._handleErrors(stream);
91 this._pipeNext(stream);
95 CombinedStream.prototype._pipeNext = function(stream) {
96 this._currentStream = stream;
98 var isStreamLike = CombinedStream.isStreamLike(stream);
100 stream.on('end', this._getNext.bind(this));
101 stream.pipe(this, {end: false});
110 CombinedStream.prototype._handleErrors = function(stream) {
112 stream.on('error', function(err) {
113 self._emitError(err);
117 CombinedStream.prototype.write = function(data) {
118 this.emit('data', data);
121 CombinedStream.prototype.pause = function() {
122 if (!this.pauseStreams) {
126 if(this.pauseStreams && this._currentStream && typeof(this._currentStream.pause) == 'function') this._currentStream.pause();
130 CombinedStream.prototype.resume = function() {
131 if (!this._released) {
132 this._released = true;
133 this.writable = true;
137 if(this.pauseStreams && this._currentStream && typeof(this._currentStream.resume) == 'function') this._currentStream.resume();
141 CombinedStream.prototype.end = function() {
146 CombinedStream.prototype.destroy = function() {
151 CombinedStream.prototype._reset = function() {
152 this.writable = false;
154 this._currentStream = null;
157 CombinedStream.prototype._checkDataSize = function() {
158 this._updateDataSize();
159 if (this.dataSize <= this.maxDataSize) {
164 'DelayedStream#maxDataSize of ' + this.maxDataSize + ' bytes exceeded.';
165 this._emitError(new Error(message));
168 CombinedStream.prototype._updateDataSize = function() {
172 this._streams.forEach(function(stream) {
173 if (!stream.dataSize) {
177 self.dataSize += stream.dataSize;
180 if (this._currentStream && this._currentStream.dataSize) {
181 this.dataSize += this._currentStream.dataSize;
185 CombinedStream.prototype._emitError = function(err) {
187 this.emit('error', err);