Race condition reading stream larger then highWaterMark

[gson]

Hi,

Found a race condition when getting fast (i.e. cached) s3 responses so that all readable event fires before _read is call.

Used following code (with some features removed, calling consume at request() and 'readable' ):

const _ = require('lodash');
const Stream = require('readable-stream');
const util = require('util');

/**
 * @constructor
 * @param {AWS.S3} client S3 client.
 * @param {Object} options AWS options.
 */
function S3ReadStream(client, options) {
  if (this instanceof S3ReadStream === false) {
    return new S3ReadStream(client, options);
  }
  if (!client || !_.isFunction(client.getObject)) {
    throw new TypeError();
  }
  if (!_.has(options, 'Bucket')) {
    throw new TypeError();
  }
  if (!_.has(options, 'Key')) {
    throw new TypeError();
  }
  Stream.Readable.call(this, _.assign({ highWaterMark: 4194304 }));
  this._more = 0;
  this.options = options;
  this.client = client;
}
util.inherits(S3ReadStream, Stream.Readable);

S3ReadStream.prototype.request = function request() {
  if (this.req) {
    return this.consume();
  }
  this.req = this.client.getObject(_.assign({}, this.options));
  this.stream = this.req
    .on('httpHeaders', statusCode => {
      if (statusCode >= 300) {
        this.emit('error', { statusCode: statusCode });
        return;
      }
    })
    .createReadStream()
    .on('end', () => {
      this.push(null);
    })
    .on('error', err => {
      this.emit('error', err);
    })
    .on('readable', () => {
      this.consume();
    });
  return this.stream;
};

S3ReadStream.prototype.consume = function consume() {
  let chunk;
  while (null !== (chunk = this.stream.read(this._more)) && this._more) {
    this._more -= chunk.length;
    this.push(chunk);
  }
};

S3ReadStream.prototype.pipe = function pipe() {
  return Stream.Readable.prototype.pipe.apply(this, arguments);
};

/**
 * @param {Number} size Amount of data to read.
 * @returns {Undefined} Nothing.
 * @see Stream.Readable._read
 */
S3ReadStream.prototype._read = function _read(size) {
  this._more += size;
  this.request();
};

module.exports = S3ReadStream;