<commit_before>'use strict';

var extend = require('util')._extend;
var util = require('util');

var EventEmitter = require('events').EventEmitter;
var Transform = require('stream').Transform;

var RRD_COOKIE = 'RRD';
var RRD_VERSION = '0003';
var FLOAT_COOKIE = 8.642135e+130;

var RRDStream = module.exports = function(options) {
  if (!(this instanceof RRDStream))
    return new RRDStream(options);

  Transform.call(this);

  this.buffer = Buffer('');
  this.rra_index = 0;

  // Default to last 24h
  this.options = extend({
    // Default to last 24h
    start: Date.now()/1000 - 86400,
    end: Date.now()/1000,
  }, options);
}

util.inherits(RRDStream, Transform);

RRDStream.prototype._transform = function(chunk, encoding, done) {
  this.buffer = Buffer.concat([this.buffer, chunk]);

  if (!this.header && this.buffer.length >= 128) {
    try {
      this.header = new RRDHeader(this.buffer.slice(0, 128));
    } catch (e) {
      this.emit('error', e);
    }
  }

  if (this.header && !this.header.ds && this.buffer.length >= this.header.header_size) {
    this.header.parse(this.buffer.slice(0, this.header.header_size));
    this.buffer = this.buffer.slice(this.header.header_size);
    this.emit('header', this.header);

    // Find the RRA that has the best coverage for the requested period.
    this.best_rra = {index:this.header.rra_cnt - 1, coverage:0};

    this.header.rra.forEach((function(rra, index) {
      var rra_end = this.header.last_update;
      var rra_start  = rra_end - (rra.rows * rra.pdp_per_row * this.header.step);

      // Out of bound
      if (this.options.end < rra_start || this.options.start > rra_end)
        return;

      rra_start = (this.options.start > rra_start) ? this.options.start : rra_start;
      rra_end = (this.options.end < rra_end) ? this.options.end : rra_end;

      var coverage = rra_end - rra_start;

      if (coverage > this.best_rra.coverage)
        this.best_rra = {index:index, coverage:coverage};
    }).bind(this));
  }

  if (!this.header || !this.header.ds)
    return done();

  // Parse each row
  while (this.buffer.length >= 8 * this.header.ds_cnt) {
    console.assert(this.rra_index <= this.best_rra.index);

    if (!this.rra) {
      this.rra = new RRDRRA(this.header, this.rra_index);

      if (this.rra_index === this.best_rra.index) {
        this.emit('rra', this.header.rra[this.rra_index]);

        this.rra.on('row', (function(timestamp, values) {
          console.assert(timestamp <= this.header.last_update);
          if (timestamp >= this.options.start && timestamp <= this.options.end) {
            this.emit('row', timestamp, values);
            this.push('{"'+timestamp+'":'+JSON.stringify(values)+'}\n');
          }
        }).bind(this));

        this.rra.on('end', (function() {
          // Ignore remaining buffer
          this.buffer.length = 0;
          this.end();
        }).bind(this));
      }<commit_msg>Fix properly clear buffer (closes #1)<commit_after>'use strict';

var extend = require('util')._extend;
var util = require('util');

var EventEmitter = require('events').EventEmitter;
var Transform = require('stream').Transform;

var RRD_COOKIE = 'RRD';
var RRD_VERSION = '0003';
var FLOAT_COOKIE = 8.642135e+130;

var RRDStream = module.exports = function(options) {
  if (!(this instanceof RRDStream))
    return new RRDStream(options);

  Transform.call(this);

  this.buffer = new Buffer(0);
  this.rra_index = 0;

  this.options = extend({
    // Default to last 24h
    start: Date.now()/1000 - 86400,
    end: Date.now()/1000,
  }, options);
}

util.inherits(RRDStream, Transform);

RRDStream.prototype._transform = function(chunk, encoding, done) {
  this.buffer = Buffer.concat([this.buffer, chunk]);

  if (!this.header && this.buffer.length >= 128) {
    try {
      this.header = new RRDHeader(this.buffer.slice(0, 128));
    } catch (e) {
      this.emit('error', e);
    }
  }

  if (this.header && !this.header.ds && this.buffer.length >= this.header.header_size) {
    this.header.parse(this.buffer.slice(0, this.header.header_size));
    this.buffer = this.buffer.slice(this.header.header_size);
    this.emit('header', this.header);

    // Find the RRA that has the best coverage for the requested period.
    this.best_rra = {index:this.header.rra_cnt - 1, coverage:0};

    this.header.rra.forEach((function(rra, index) {
      var rra_end = this.header.last_update;
      var rra_start  = rra_end - (rra.rows * rra.pdp_per_row * this.header.step);

      // Out of bound
      if (this.options.end < rra_start || this.options.start > rra_end)
        return;

      rra_start = (this.options.start > rra_start) ? this.options.start : rra_start;
      rra_end = (this.options.end < rra_end) ? this.options.end : rra_end;

      var coverage = rra_end - rra_start;

      if (coverage > this.best_rra.coverage)
        this.best_rra = {index:index, coverage:coverage};
    }).bind(this));
  }

  if (!this.header || !this.header.ds)
    return done();

  // Parse each row
  while (this.buffer && this.buffer.length >= 8 * this.header.ds_cnt) {
    console.assert(this.rra_index <= this.best_rra.index);

    if (!this.rra) {
      this.rra = new RRDRRA(this.header, this.rra_index);

      if (this.rra_index === this.best_rra.index) {
        this.emit('rra', this.header.rra[this.rra_index]);

        this.rra.on('row', (function(timestamp, values) {
          console.assert(timestamp <= this.header.last_update);
          if (timestamp >= this.options.start && timestamp <= this.options.end) {
            this.emit('row', timestamp, values);
            this.push('{"'+timestamp+'":'+JSON.stringify(values)+'}\n');
          }
        }).bind(this));

        this.rra.on('end', (function() {
          // Ignore remaining buffer
          this.buffer = new Buffer(0);
          this.end();
        }).bind(this));
      }
{'fraction_non_alphanumeric': 0.15906771010547804, 'fraction_numerical': 0.01497107859816264, 'mean_word_length': 2.927187708750835, 'pattern_counts': {'":': 2, '<': 13, '<?xml version=': 0, '>': 17, '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': '3642163', 'n_tokens_mistral': 2084, 'n_tokens_neox': 1980, 'n_words': 392}