combined_stream.js 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190
  1. var util = require('util');
  2. var Stream = require('stream').Stream;
  3. var DelayedStream = require('delayed-stream');
  4. var defer = require('./defer.js');
  5. module.exports = CombinedStream;
  6. function CombinedStream() {
  7. this.writable = false;
  8. this.readable = true;
  9. this.dataSize = 0;
  10. this.maxDataSize = 2 * 1024 * 1024;
  11. this.pauseStreams = true;
  12. this._released = false;
  13. this._streams = [];
  14. this._currentStream = null;
  15. }
  16. util.inherits(CombinedStream, Stream);
  17. CombinedStream.create = function(options) {
  18. var combinedStream = new this();
  19. options = options || {};
  20. for (var option in options) {
  21. combinedStream[option] = options[option];
  22. }
  23. return combinedStream;
  24. };
  25. CombinedStream.isStreamLike = function(stream) {
  26. return (typeof stream !== 'function')
  27. && (typeof stream !== 'string')
  28. && (typeof stream !== 'boolean')
  29. && (typeof stream !== 'number')
  30. && (!Buffer.isBuffer(stream));
  31. };
  32. CombinedStream.prototype.append = function(stream) {
  33. var isStreamLike = CombinedStream.isStreamLike(stream);
  34. if (isStreamLike) {
  35. if (!(stream instanceof DelayedStream)) {
  36. var newStream = DelayedStream.create(stream, {
  37. maxDataSize: Infinity,
  38. pauseStream: this.pauseStreams,
  39. });
  40. stream.on('data', this._checkDataSize.bind(this));
  41. stream = newStream;
  42. }
  43. this._handleErrors(stream);
  44. if (this.pauseStreams) {
  45. stream.pause();
  46. }
  47. }
  48. this._streams.push(stream);
  49. return this;
  50. };
  51. CombinedStream.prototype.pipe = function(dest, options) {
  52. Stream.prototype.pipe.call(this, dest, options);
  53. this.resume();
  54. return dest;
  55. };
  56. CombinedStream.prototype._getNext = function() {
  57. this._currentStream = null;
  58. var stream = this._streams.shift();
  59. if (typeof stream == 'undefined') {
  60. this.end();
  61. return;
  62. }
  63. if (typeof stream !== 'function') {
  64. this._pipeNext(stream);
  65. return;
  66. }
  67. var getStream = stream;
  68. getStream(function(stream) {
  69. var isStreamLike = CombinedStream.isStreamLike(stream);
  70. if (isStreamLike) {
  71. stream.on('data', this._checkDataSize.bind(this));
  72. this._handleErrors(stream);
  73. }
  74. defer(this._pipeNext.bind(this, stream));
  75. }.bind(this));
  76. };
  77. CombinedStream.prototype._pipeNext = function(stream) {
  78. this._currentStream = stream;
  79. var isStreamLike = CombinedStream.isStreamLike(stream);
  80. if (isStreamLike) {
  81. stream.on('end', this._getNext.bind(this));
  82. stream.pipe(this, {end: false});
  83. return;
  84. }
  85. var value = stream;
  86. this.write(value);
  87. this._getNext();
  88. };
  89. CombinedStream.prototype._handleErrors = function(stream) {
  90. var self = this;
  91. stream.on('error', function(err) {
  92. self._emitError(err);
  93. });
  94. };
  95. CombinedStream.prototype.write = function(data) {
  96. this.emit('data', data);
  97. };
  98. CombinedStream.prototype.pause = function() {
  99. if (!this.pauseStreams) {
  100. return;
  101. }
  102. if(this.pauseStreams && this._currentStream && typeof(this._currentStream.pause) == 'function') this._currentStream.pause();
  103. this.emit('pause');
  104. };
  105. CombinedStream.prototype.resume = function() {
  106. if (!this._released) {
  107. this._released = true;
  108. this.writable = true;
  109. this._getNext();
  110. }
  111. if(this.pauseStreams && this._currentStream && typeof(this._currentStream.resume) == 'function') this._currentStream.resume();
  112. this.emit('resume');
  113. };
  114. CombinedStream.prototype.end = function() {
  115. this._reset();
  116. this.emit('end');
  117. };
  118. CombinedStream.prototype.destroy = function() {
  119. this._reset();
  120. this.emit('close');
  121. };
  122. CombinedStream.prototype._reset = function() {
  123. this.writable = false;
  124. this._streams = [];
  125. this._currentStream = null;
  126. };
  127. CombinedStream.prototype._checkDataSize = function() {
  128. this._updateDataSize();
  129. if (this.dataSize <= this.maxDataSize) {
  130. return;
  131. }
  132. var message =
  133. 'DelayedStream#maxDataSize of ' + this.maxDataSize + ' bytes exceeded.';
  134. this._emitError(new Error(message));
  135. };
  136. CombinedStream.prototype._updateDataSize = function() {
  137. this.dataSize = 0;
  138. var self = this;
  139. this._streams.forEach(function(stream) {
  140. if (!stream.dataSize) {
  141. return;
  142. }
  143. self.dataSize += stream.dataSize;
  144. });
  145. if (this._currentStream && this._currentStream.dataSize) {
  146. this.dataSize += this._currentStream.dataSize;
  147. }
  148. };
  149. CombinedStream.prototype._emitError = function(err) {
  150. this._reset();
  151. this.emit('error', err);
  152. };