<commit_before>const Writable = require('stream').Writable;

class Reader {
  constructor(input, opt) {
    this._helper = new StreamReaderHelper(input, opt);
  }

  *read(encoding) {
    const chunk = yield this._helper.readData();
    return chunk && (encoding ? chunk.toString(encoding) : chunk);
  }
}

class LineReader {
  constructor(input) {
    this._helper = new StreamReaderHelper(input);
    this._lines = [];
  }

  *read() {
    if (this._lines.length > 0) {
      return this._lines.shift().replace('\r', '');
    } else {
      while (this._lines.length === 0) {
        let chunk = yield this._helper.readData();
        if (!chunk) {
          let ret = this._incomplete;
          this._incomplete = null;
          return ret;
        }

        let data = chunk.toString(),
            lines = data.split('\n'),
            incomplete = !data.endsWith('\n');

        if (!incomplete) lines.pop();
        if (this._incomplete) lines[0] = this._incomplete + lines[0];

        this._incomplete = incomplete ? lines.pop() : null;

        this._lines = lines;
      }

      return this._lines.shift().replace('\r', '');
    }
  }
}

module.exports = { Reader, LineReader };

class StreamReaderHelper extends Writable {
  constructor(input, opt) {
    super(opt);

    let me = this;
    input.pipe(this);
    input.on('error', function (err) {
      me._err = err;
      me._callReadCb(err);
    });

    this.on('finish', function () {
      me._end = true;
      me._callReadCb();
    });
  }

  _callReadCb() {
    let rcb = this._readCb;
    if (rcb) {
      this._readCb = null;

      let args = arguments;
      process.nextTick(function () { rcb.apply(null, args); });
    }
  }

  readData() {
    return cb => {
      if (this._readCb) return cb(new Error('Reader already exists.'));

      if (this._data) {
        let data = this._data,
          dataCb = this._dataCb;

        this._data = null;
        this._dataCb = null;

        process.nextTick(function () {
          dataCb();
          cb(null, data);
        });
      } else if (this._end) {
        process.nextTick(cb);
      } else if (this._err) {
        process.nextTick(function () { cb(this._err); });
      } else {
        this._readCb = cb;
      }
    };
  }

  _write(chunk, encoding, callback) {
    if (this._readCb) {
      this._callReadCb(null, chunk);
      callback();
    } else {
      this._data = chunk;
      this._dataCb = callback;
    }
  }
}<commit_msg>use es6 style.<commit_after>const Writable = require('stream').Writable;

class Reader {
  constructor(input, opt) {
    this._helper = new StreamReaderHelper(input, opt);
  }

  *read(encoding) {
    const chunk = yield this._helper.readData();
    return chunk && (encoding ? chunk.toString(encoding) : chunk);
  }
}

class LineReader {
  constructor(input) {
    this._helper = new StreamReaderHelper(input);
    this._lines = [];
  }

  *read() {
    if (this._lines.length > 0) {
      return this._lines.shift().replace('\r', '');
    } else {
      while (this._lines.length === 0) {
        let chunk = yield this._helper.readData();
        if (!chunk) {
          let ret = this._incomplete;
          this._incomplete = null;
          return ret;
        }

        let data = chunk.toString(),
            lines = data.split('\n'),
            incomplete = !data.endsWith('\n');

        if (!incomplete) lines.pop();
        if (this._incomplete) lines[0] = this._incomplete + lines[0];

        this._incomplete = incomplete ? lines.pop() : null;

        this._lines = lines;
      }

      return this._lines.shift().replace('\r', '');
    }
  }
}

module.exports = { Reader, LineReader };

class StreamReaderHelper extends Writable {
  constructor(input, opt) {
    super(opt);

    input.pipe(this);
    input.on('error', err => {
      this._err = err;
      this._callReadCb(err);
    });

    this.on('finish', () => {
      this._end = true;
      this._callReadCb();
    });
  }

  _callReadCb() {
    let rcb = this._readCb;
    if (rcb) {
      this._readCb = null;

      let args = arguments;
      process.nextTick(function () { rcb.apply(null, args); });
    }
  }

  readData() {
    return cb => {
      if (this._readCb) return cb(new Error('Reader already exists.'));

      if (this._data) {
        let data = this._data,
            dataCb = this._dataCb;

        this._data = null;
        this._dataCb = null;

        process.nextTick(function () {
          dataCb();
          cb(null, data);
        });
      } else if (this._end) {
        process.nextTick(cb);
      } else if (this._err) {
        process.nextTick(() => cb(this._err));
      } else {
        this._readCb = cb;
      }
    };
  }

  _write(chunk, encoding, callback) {
    if (this._readCb) {
      this._callReadCb(null, chunk);
      callback();
    } else {
      this._data = chunk;
      this._dataCb = callback;
    }
  }
}
{'fraction_non_alphanumeric': 0.1531586431037985, 'fraction_numerical': 0.0018281535648994515, 'mean_word_length': 2.047029702970297, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 10, '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': '2168889', 'n_tokens_mistral': 1612, 'n_tokens_neox': 1576, 'n_words': 323}