<commit_before>'use strict';

var stream = require('stream');
var fs = require('fs');
var path = require('path');
var util = require('util');

var jsu = require('./JSutils');
var fm = require('./daFileManager');
var conf = require('../velstor-davcnc.conf');

var max_buf_size = conf.MaxBufferSize();

/**
 * This class is a custom write stream to output VDA messages to multiple files
 * to date names directories
 */
function DA_WriteFileStream(root, pfx, sfx)
{
    var self = this;
    self._wstream = null;
    self._buf_size = 0;
    self._file_size = 0;
    self._curr_file = "";            ///< Current file
    self._last_record = 0;
//    console.log('DA_WriteFileStream: prefix='+ pfx)
    self._stockBuffer = new Buffer(''); // empty
    self._fileMgr = new fm.DAFileManager(root, pfx, sfx);
    self.CreateWriteStream();
    stream.Writable.call(this, {objectMode: true, highWaterMark: max_buf_size});
}

util.inherits(DA_WriteFileStream, stream.Writable);

DA_WriteFileStream.prototype.CreateWriteStream = function()
{
    var self = this;
    self._file_size = 0;
    self._last_record = Date.now();
    self._curr_file = self._fileMgr.GetPath();
    console.log('DA_WriteFileStream.prototype.CreateWriteStream: logFile=' + self._curr_file);
    self._wstream = fs.createWriteStream(self._curr_file);
//    self._wstream.cork();
}

// DA_WriteFileStream.prototype._write = function(data, encoding, callback)
DA_WriteFileStream.prototype._write = function(data, encoding, cb)
{
    var self = this;
    self._last_record = Date.now();
//    console.log('Ready to write');
    // Check buffer is full or expired
    //
    if (self._fileMgr.ValidStream(self._buf_size) === false) {
//        console.log('Write buffer: ' + self._stockBuffer.length);
        self._fileStat();
        self._flush();
    }

    // Check file size exceeded max size or date changed (It's time to create a new file)
    //
    if (self._fileMgr.ChangeDate() !== true) {
        console.log('Create a new date directory');
        self._flush();
        self.CreateWriteStream();
    } else if (self._file_size > 0 && true !== self._fileMgr.ValidFile(self._file_size)) {
        console.log('Create a new file');
        self.CreateWriteStream();
    }

    self._buf_size += data.length;

    var buffer = new Buffer(data);
    self._stockBuffer = Buffer.concat([self._stockBuffer, buffer]);
//        console.log('Added new record: _stockBuffer=' + self._stockBuffer.length + ' buf_size=' + self._buf_size
//                     + ' data.length=' + data.length + ' buffer.length=' + buffer.length);
//        console.log('StockBuffer=' + stockBuffer.toString())

}

// Flush stockBuffer
//
DA_WriteFileStream.prototype._flush = function(data, encoding, cb)
{
    var self = this;
    if (self._stockBuffer.length > 0) {
        self._wstream.write(self._stockBuffer);
        self._file_size += self._stockBuffer.length;
        self._stockBuffer = new Buffer('');
        self._buf_size = 0;
    }
}

// Timeout flush stockBuffer
//
DA_WriteFileStream.prototype._time_flush = function()
{
    var self = this;
    var curr_timeout = Date.now() - self._last_record;
    var timeout = conf.FlushTimeout();
    if (curr_timeout > timeout) {
        self._flush();
        self._last_record = Date.now();
    }
}<commit_msg>Fixed bug<commit_after>'use strict';

var stream = require('stream');
var fs = require('fs');
var path = require('path');
var util = require('util');

var jsu = require('./JSutils');
var fm = require('./daFileManager');
var conf = require('../velstor-davcnc.conf');

var max_buf_size = conf.MaxBufferSize();

/**
 * This class is a custom write stream to output VDA messages to multiple files
 * to date names directories
 */
function DA_WriteFileStream(root, pfx, sfx)
{
    var self = this;
    self._wstream = null;
    self._buf_size = 0;
    self._file_size = 0;
    self._curr_file = "";            ///< Current file
    self._last_record = 0;
//    console.log('DA_WriteFileStream: prefix='+ pfx)
    self._stockBuffer = new Buffer(''); // empty
    self._fileMgr = new fm.DAFileManager(root, pfx, sfx);
    self.CreateWriteStream();
    stream.Writable.call(this, {objectMode: true, highWaterMark: max_buf_size});
}

util.inherits(DA_WriteFileStream, stream.Writable);

DA_WriteFileStream.prototype.CreateWriteStream = function()
{
    var self = this;
    self._file_size = 0;
    self._last_record = Date.now();
    self._curr_file = self._fileMgr.GetPath();
    console.log('DA_WriteFileStream.prototype.CreateWriteStream: logFile=' + self._curr_file);
    self._wstream = fs.createWriteStream(self._curr_file);
//    self._wstream.cork();
}

// DA_WriteFileStream.prototype._write = function(data, encoding, callback)
DA_WriteFileStream.prototype._write = function(data, encoding, cb)
{
    var self = this;
    self._last_record = Date.now();
//    console.log('Ready to write');
    // Check buffer is full or expired
    //
    if (self._fileMgr.ValidStream(self._buf_size) === false) {
//        console.log('Write buffer: ' + self._stockBuffer.length);
        self._fileStat();
        self.flush();
    }

    // Check file size exceeded max size or date changed (It's time to create a new file)
    //
    if (self._fileMgr.ChangeDate() !== true) {
        console.log('Create a new date directory');
        self._flush();
        self.CreateWriteStream();
    } else if (self._file_size > 0 && true !== self._fileMgr.ValidFile(self._file_size)) {
        console.log('Create a new file');
        self.CreateWriteStream();
    }

    self._buf_size += data.length;

    var buffer = new Buffer(data);
    self._stockBuffer = Buffer.concat([self._stockBuffer, buffer]);
//        console.log('Added new record: _stockBuffer=' + self._stockBuffer.length + ' buf_size=' + self._buf_size
//                     + ' data.length=' + data.length + ' buffer.length=' + buffer.length);
//        console.log('StockBuffer=' + stockBuffer.toString())

}

// Flush stockBuffer
//
DA_WriteFileStream.prototype.flush = function(data, encoding, cb)
{
    var self = this;
    if (self._stockBuffer.length > 0) {
        self._wstream.write(self._stockBuffer);
        self._file_size += self._stockBuffer.length;
        self._stockBuffer = new Buffer('');
        self._buf_size = 0;
    }
}

// Timeout flush stockBuffer
//
DA_WriteFileStream.prototype.time_flush = function()
{
    var self = this;
    var curr_timeout = Date.now() - self._last_record;
    var timeout = conf.FlushTimeout();
    if (curr_timeout > timeout) {
        self.flush();
        self._last_record = Date.now();
    }
}
{'fraction_non_alphanumeric': 0.14350453172205438, 'fraction_numerical': 0.0021148036253776435, 'mean_word_length': 3.0971534653465347, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, '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': '7054990', 'n_tokens_mistral': 2072, 'n_tokens_neox': 2044, 'n_words': 459}