combined_stream.js 4.6 KB

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