1/*
  2 * buffered-stream.js: A simple(r) Stream which is partially buffered into memory.
  3 *
  4 * (C) 2010, Mikeal Rogers
  5 *
  6 * Adapted for Flatiron
  7 * (C) 2011, Charlie Robbins & the Contributors
  8 * MIT LICENSE
  9 *
 10 */
 11
 12var events = require('events'),
 13    fs = require('fs'),
 14    stream = require('stream'),
 15    util = require('util');
 16
 17//
 18// ### function BufferedStream (limit)
 19// #### @limit {number} **Optional** Size of the buffer to limit
 20// Constructor function for the BufferedStream object responsible for
 21// maintaining a stream interface which can also persist to memory
 22// temporarily.
 23//
 24
 25var BufferedStream = module.exports = function (limit) {
 26  events.EventEmitter.call(this);
 27
 28  if (typeof limit === 'undefined') {
 29    limit = Infinity;
 30  }
 31
 32  this.limit = limit;
 33  this.size = 0;
 34  this.chunks = [];
 35  this.writable = true;
 36  this.readable = true;
 37  this._buffer = true;
 38};
 39
 40util.inherits(BufferedStream, stream.Stream);
 41
 42Object.defineProperty(BufferedStream.prototype, 'buffer', {
 43  get: function () {
 44    return this._buffer;
 45  },
 46  set: function (value) {
 47    if (!value && this.chunks) {
 48      var self = this;
 49      this.chunks.forEach(function (c) { self.emit('data', c) });
 50      if (this.ended) this.emit('end');
 51      this.size = 0;
 52      delete this.chunks;
 53    }
 54
 55    this._buffer = value;
 56  }
 57});
 58
 59BufferedStream.prototype.pipe = function () {
 60  var self = this,
 61      dest;
 62
 63  if (self.resume) {
 64    self.resume();
 65  }
 66
 67  dest = stream.Stream.prototype.pipe.apply(self, arguments);
 68
 69  //
 70  // just incase you are piping to two streams, do not emit data twice.
 71  // note: you can pipe twice, but you need to pipe both streams in the same tick.
 72  // (this is normal for streams)
 73  //
 74  if (this.piped) {
 75    return dest;
 76  }
 77
 78  process.nextTick(function () {
 79    if (self.chunks) {
 80      self.chunks.forEach(function (c) { self.emit('data', c) });
 81      self.size = 0;
 82      delete self.chunks;
 83    }
 84
 85    if (!self.readable) {
 86      if (self.ended) {
 87        self.emit('end');
 88      }
 89      else if (self.closed) {
 90        self.emit('close');
 91      }
 92    }
 93  });
 94
 95  this.piped = true;
 96
 97  return dest;
 98};
 99
100BufferedStream.prototype.write = function (chunk) {
101  if (!this.chunks || this.piped) {
102    this.emit('data', chunk);
103    return;
104  }
105
106  this.chunks.push(chunk);
107  this.size += chunk.length;
108  if (this.limit < this.size) {
109    this.pause();
110  }
111};
112
113BufferedStream.prototype.end = function () {
114  this.readable = false;
115  this.ended = true;
116  this.emit('end');
117};
118
119BufferedStream.prototype.destroy = function () {
120  this.readable = false;
121  this.writable = false;
122  delete this.chunks;
123};
124
125BufferedStream.prototype.close = function () {
126  this.readable = false;
127  this.closed = true;
128};
129
130if (!stream.Stream.prototype.pause) {
131  BufferedStream.prototype.pause = function () {
132    this.emit('pause');
133  };
134}
135
136if (!stream.Stream.prototype.resume) {
137  BufferedStream.prototype.resume = function () {
138    this.emit('resume');
139  };
140}
141