node_modules/union/lib/buffered-stream.js

/*
 * buffered-stream.js: A simple(r) Stream which is partially buffered into memory.
 *
 * (C) 2010, Mikeal Rogers
 *
 * Adapted for Flatiron
 * (C) 2011, Charlie Robbins & the Contributors
 * MIT LICENSE
 *
 */

var events = require('events'),
    fs = require('fs'),
    stream = require('stream'),
    util = require('util');

//
// ### function BufferedStream (limit)
// #### @limit {number} **Optional** Size of the buffer to limit
// Constructor function for the BufferedStream object responsible for
// maintaining a stream interface which can also persist to memory
// temporarily.
//

var BufferedStream = module.exports = function (limit) {
  events.EventEmitter.call(this);

  if (typeof limit === 'undefined') {
    limit = Infinity;
  }

  this.limit = limit;
  this.size = 0;
  this.chunks = [];
  this.writable = true;
  this.readable = true;
  this._buffer = true;
};

util.inherits(BufferedStream, stream.Stream);

Object.defineProperty(BufferedStream.prototype, 'buffer', {
  get: function () {
    return this._buffer;
  },
  set: function (value) {
    if (!value && this.chunks) {
      var self = this;
      this.chunks.forEach(function (c) { self.emit('data', c) });
      if (this.ended) this.emit('end');
      this.size = 0;
      delete this.chunks;
    }

    this._buffer = value;
  }
});

BufferedStream.prototype.pipe = function () {
  var self = this,
      dest;

  if (self.resume) {
    self.resume();
  }

  dest = stream.Stream.prototype.pipe.apply(self, arguments);

  //
  // just incase you are piping to two streams, do not emit data twice.
  // note: you can pipe twice, but you need to pipe both streams in the same tick.
  // (this is normal for streams)
  //
  if (this.piped) {
    return dest;
  }

  process.nextTick(function () {
    if (self.chunks) {
      self.chunks.forEach(function (c) { self.emit('data', c) });
      self.size = 0;
      delete self.chunks;
    }

    if (!self.readable) {
      if (self.ended) {
        self.emit('end');
      }
      else if (self.closed) {
        self.emit('close');
      }
    }
  });

  this.piped = true;

  return dest;
};

BufferedStream.prototype.write = function (chunk) {
  if (!this.chunks || this.piped) {
    this.emit('data', chunk);
    return;
  }

  this.chunks.push(chunk);
  this.size += chunk.length;
  if (this.limit < this.size) {
    this.pause();
  }
};

BufferedStream.prototype.end = function () {
  this.readable = false;
  this.ended = true;
  this.emit('end');
};

BufferedStream.prototype.destroy = function () {
  this.readable = false;
  this.writable = false;
  delete this.chunks;
};

BufferedStream.prototype.close = function () {
  this.readable = false;
  this.closed = true;
};

if (!stream.Stream.prototype.pause) {
  BufferedStream.prototype.pause = function () {
    this.emit('pause');
  };
}

if (!stream.Stream.prototype.resume) {
  BufferedStream.prototype.resume = function () {
    this.emit('resume');
  };
}