<commit_before>"use strict";
var Readable = require('stream').Readable;
var Cursor = require('./cursor.js');
var errors = require('./error');
var util = require('util');

// Experimental, but should work fine.
function ReadableStream(options, cursor) {
  if (cursor) this._cursor = cursor;
  this._pending = 0; // How many time we called _read while no cursor was available
  this._index = 0;
  this._recursion = 0;
  this._maxRecursion = 1000; // Hardcoded
  this._highWaterMark = options.highWaterMark;

  Readable.call(this, {
    objectMode: true,
    highWaterMark: this._highWaterMark
  });
}
util.inherits(ReadableStream, Readable);

ReadableStream.prototype._setCursor = function(cursor) {
  if (cursor instanceof Cursor === false) {
    this.emit('error', new Error('Cannot create a stream on a single value.'));
    return this;
  }

  this._cursor = cursor;
  this._fetch({ decrement: true });
};

ReadableStream.prototype._read = function(size) {
  this._count++;
  if (this._cursor === undefined) {
    this._pending++;
    return;
  }

  this._recursion = 0;
  this._fetch();
};

ReadableStream.prototype._fetch = function(options) {
  var self = this;
  options = options || {};
  if (!!options.decrement) {
    self._pending--;
    if (self._pending < 0) return;
  }

  if (self._cursor._closed === true) {
    self.push(null);
    return;
  }

  var continueFetching = function() {
    if (self._recursion++ === self._maxRecursion) {
      process.nextTick(function() { self._fetch(options); });
    } else {
      self._fetch(options);
    }
  };

  self._cursor._next().then(function(data) {
    // Silently drop null values for now
    if (data === null) return continueFetching();

    // otherwise push the data, and continue if successful
    if (self.push(data)) return continueFetching();
  })
  .catch(errors.ReqlBaseError, function(error) {
    if (error.message.match(/No more rows in the/)) {
      self.push(null);
    } else if (error.message === 'You cannot retrieve data from a cursor that is closed.') {
      // if the user call `close`, the cursor may reject pending requests. We just
      // ignore them here.
    } else {
      self.emit('error', error);
    }
  });
};

ReadableStream.prototype.close = function() {
  this.push(null);<commit_msg>refactor(stream): track if stream is manually closed<commit_after>"use strict";
var Readable = require('stream').Readable;
var Cursor = require('./cursor.js');
var errors = require('./error');
var util = require('util');

// Experimental, but should work fine.
function ReadableStream(options, cursor) {
  if (cursor) this._cursor = cursor;
  this._pending = 0; // How many time we called _read while no cursor was available
  this._index = 0;
  this._recursion = 0;
  this._maxRecursion = 1000; // Hardcoded
  this._highWaterMark = options.highWaterMark;
  this._closed = false;

  Readable.call(this, {
    objectMode: true,
    highWaterMark: this._highWaterMark
  });
}
util.inherits(ReadableStream, Readable);

ReadableStream.prototype._setCursor = function(cursor) {
  if (cursor instanceof Cursor === false) {
    this.emit('error', new Error('Cannot create a stream on a single value.'));
    return this;
  }

  this._cursor = cursor;
  this._fetch({ decrement: true });
};

ReadableStream.prototype._read = function(size) {
  this._count++;
  if (this._cursor === undefined) {
    this._pending++;
    return;
  }

  this._recursion = 0;
  this._fetch();
};

ReadableStream.prototype._fetch = function(options) {
  if (this._closed === true) return;

  var self = this;
  options = options || {};
  if (!!options.decrement) {
    self._pending--;
    if (self._pending < 0) return;
  }

  if (self._cursor._closed === true) {
    self.push(null);
    return;
  }

  var continueFetching = function() {
    if (self._recursion++ === self._maxRecursion) {
      process.nextTick(function() { self._fetch(options); });
    } else {
      self._fetch(options);
    }
  };

  self._cursor._next().then(function(data) {
    if (self._closed === true) return;

    // Silently drop null values for now
    if (data === null) return continueFetching();

    // otherwise push the data, and continue if successful
    if (self.push(data)) return continueFetching();
  })
  .catch(errors.ReqlBaseError, function(error) {
    if (error.message.match(/No more rows in the/)) {
      self.push(null);
    } else if (error.message === 'You cannot retrieve data from a cursor that is closed.') {
      // if the user call `close`, the cursor may reject pending requests. We just
      // ignore them here.
    } else {
      self.emit('error', error);
    }
  });
};

ReadableStream.prototype.close = function() {
  if (this._closed) return;

  this.push(null);
  this._closed = true;
{'fraction_non_alphanumeric': 0.14501573976915005, 'fraction_numerical': 0.003777544596012592, 'mean_word_length': 3.3170289855072466, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, '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': '1144884', 'n_tokens_mistral': 1492, 'n_tokens_neox': 1452, 'n_words': 396}