<commit_before>
/**
 * Module dependencies
 */

var util = require('util');
var PassThrough  = require('stream').PassThrough;

/**
 * expose
 * @ignore
 */

module.exports = exports = GridReadStream;

/**
 * GridReadStream
 *
 * @param {Grid} grid
 * @param {Object} options
 */

function GridReadStream (grid, options) {
  if (!(this instanceof GridReadStream))
    return new GridReadStream(grid, options);

  PassThrough.call(this);
  this._opened = false;
  this._opening = false;

  this._grid = grid;

  // a bit backwards compatible
  if (typeof options === 'string') {
    options = { filename: options };
  }

  this.options = options || {};

  if(options._id) {
    this.id = grid.tryParseObjectId(options._id);

    if(!this.id) {
      this.id = options._id;
    }
  }

  this.name = this.options.filename || '';
  this.mode = 'r';

  this.range = this.options.range || { startPos: 0, endPos: undefined };
  if (typeof(this.range.startPos) === 'undefined') {
    this.range.startPos = 0;
  }

  this._store = new grid.mongo.GridStore(grid.db, this.id || new grid.mongo.ObjectID(), this.name, this.mode, this.options);
  // Workaround for Gridstore issue https://github.com/mongodb/node-mongodb-native/pull/930
  if (!this.id) {
    //var REFERENCE_BY_FILENAME = 0,
    this._store.referenceBy = 0;
  }

  this._open();
}

/**
 * Inherit from stream.PassThrough
 * @ignore
 */

util.inherits(GridReadStream, PassThrough);

/**
 * _open
 *
 * @api private
 */

GridReadStream.prototype._open = function _open () {
  if (this._opening) return;
  this._opening = true;

  var self = this;
  this._store.open(function (err, gs) {
    if (err) return self._error(err);
    self._opened = true;
    gs.seek(self.range.startPos, function(err) {
      if (err) return self._error(err);
      self.emit('open');
      self._pipe(gs.stream());
    });
  });
}

/**
 * _pipe
 *
 * @api private
 */

GridReadStream.prototype._pipe = function _pipe (gsReadStream) {
  var self = this;
  if (!this._opened) return self._error('Unable to pipe. Expected gridstore to be open');

  //Init range request
  var currentPos = self.range.startPos;
  var isGreatThanEndPos = false;

  //Pipe implementation
  //Unable to just use gsReadStream.pipe(self) becuase you cant sepeficy end range in gridStore, only where to begin.
  //No need to handle setEncoding as the PassThrough stream handels it.
  gsReadStream.on('data', function (data) {
    //Control the range of data sent
    if (typeof(self.range.endPos) !== 'undefined' && currentPos + data.length > self.range.endPos + 1) {
      isGreatThanEndPos = true;
      data = data.slice(0, self.range.endPos - currentPos + 1);
    }

    //Write data and pause if stream gets saturated
    var ready = self.write(data);
    if (ready === false) {
        gsReadStream.pause();
        self.once('drain', this.resume.bind(this));
    }

    //If end of requested range then end stream and close the store
    if (isGreatThanEndPos) {
      gsReadStream.end();
      self.end();
      self._close();
    }

    //Keep track of the data that is sent
    currentPos += data.length;
  });

  //If end of stream is reached emit `end` and close the store
  gsReadStream.on('end', function () {
    if (!isGreatThanEndPos) {
        self.end();
        self._close();<commit_msg>Pass on error events from gfReadStream, handle end from passthrough readstream.<commit_after>
/**
 * Module dependencies
 */

var util = require('util');
var PassThrough  = require('stream').PassThrough;

/**
 * expose
 * @ignore
 */

module.exports = exports = GridReadStream;

/**
 * GridReadStream
 *
 * @param {Grid} grid
 * @param {Object} options
 */

function GridReadStream (grid, options) {
  if (!(this instanceof GridReadStream))
    return new GridReadStream(grid, options);

  PassThrough.call(this);
  this._opened = false;
  this._opening = false;
  this._end = false;

  this._grid = grid;

  // a bit backwards compatible
  if (typeof options === 'string') {
    options = { filename: options };
  }

  this.options = options || {};

  if(options._id) {
    this.id = grid.tryParseObjectId(options._id);

    if(!this.id) {
      this.id = options._id;
    }
  }

  this.name = this.options.filename || '';
  this.mode = 'r';

  this.range = this.options.range || { startPos: 0, endPos: undefined };
  if (typeof(this.range.startPos) === 'undefined') {
    this.range.startPos = 0;
  }

  this._store = new grid.mongo.GridStore(grid.db, this.id || new grid.mongo.ObjectID(), this.name, this.mode, this.options);
  // Workaround for Gridstore issue https://github.com/mongodb/node-mongodb-native/pull/930
  if (!this.id) {
    // var REFERENCE_BY_FILENAME = 0,
    this._store.referenceBy = 0;
  }

  this._open();
}

/**
 * Inherit from stream.PassThrough
 * @ignore
 */

util.inherits(GridReadStream, PassThrough);

/**
 * _open
 *
 * @api private
 */

GridReadStream.prototype._open = function _open () {
  if (this._opening) return;
  this._opening = true;

  var self = this;
  this._store.open(function (err, gs) {
    if (err) return self._error(err);
    self._opened = true;
    gs.seek(self.range.startPos, function(err) {
      if (err) return self._error(err);
      self.emit('open');
      self._pipe(gs.stream());
    });
  });
}

/**
 * _pipe
 *
 * @api private
 */

GridReadStream.prototype._pipe = function _pipe (gsReadStream) {
  var self = this;
  if (!this._opened) return self._error('Unable to pipe. Expected gridstore to be open');

  // Init range request
  var currentPos = self.range.startPos;
  var isGreatThanEndPos = false;

  // Pipe implementation
  // Unable to just use gsReadStream.pipe(self) becuase you cant sepeficy end range in gridStore, only where to begin.
  // No need to handle setEncoding as the PassThrough stream handels it.
  gsReadStream.on('data', function (data) {
    // Control the range of data sent
    if (typeof(self.range.endPos) !== 'undefined' && currentPos + data.length > self.range.endPos + 1) {
      isGreatThanEndPos = true;
      data = data.slice(0, self.range.endPos - currentPos + 1);
    }

    // Write data and pause if stream gets saturated
    var ready = self.write(data);
    if (ready === false) {
        gsReadStream.pause();
        self.once('drain', this.resume.bind(this));
    }

    // If end of requested range then end stream and close the store
    if (isGreatThanEndPos) {
      self._end = true;
      gsReadStream.end();
      self.end();
      self._close();
    }

    // Keep track of the data that is sent
    currentPos += data.length;
  });

  // If end of gsReadStream is reached emit `end` and close the store
  gsReadStream.on('end', function () {
    if (!isGreatThanEndPos && !self._end) {
        self._end = true;
        self.end();
        self._close();
    }
  });

  // Pass on error events and clsoe the store
  gsReadStream.on('error', function(err) {
    self._error(err);
  });

  // If stream was ended end gsReadStream and close the store
  self.on('end', function() {
    if (!self._end) {
      self._end = true;
      gsReadStream.end();
      self._close();
{'fraction_non_alphanumeric': 0.14243490499648134, 'fraction_numerical': 0.0028149190710767065, 'mean_word_length': 2.925966850828729, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4466027', 'n_tokens_mistral': 2373, 'n_tokens_neox': 2275, 'n_words': 601}