node_modules/streamroller/lib/RollingFileWriteStream.js - nifi-fds - Git at Google

 const debug = require("debug")("streamroller:RollingFileWriteStream");
 const fs = require("fs-extra");
 const path = require("path");
 const newNow = require("./now");
 const format = require("date-format");
 const { Writable } = require("stream");
 const fileNameFormatter = require("./fileNameFormatter");
 const fileNameParser = require("./fileNameParser");
 const moveAndMaybeCompressFile = require("./moveAndMaybeCompressFile");

 /**
  * RollingFileWriteStream is mainly used when writing to a file rolling by date or size.
  * RollingFileWriteStream inherits from stream.Writable
  */
 class RollingFileWriteStream extends Writable {
   /**
    * Create a RollingFileWriteStream
    * @constructor
    * @param {string} filePath - The file path to write.
    * @param {object} options - The extra options
    * @param {number} options.numToKeep - The max numbers of files to keep.
    * @param {number} options.maxSize - The maxSize one file can reach. Unit is Byte.
    *                                   This should be more than 1024. The default is Number.MAX_SAFE_INTEGER.
    * @param {string} options.mode - The mode of the files. The default is '0644'. Refer to stream.writable for more.
    * @param {string} options.flags - The default is 'a'. Refer to stream.flags for more.
    * @param {boolean} options.compress - Whether to compress backup files.
    * @param {boolean} options.keepFileExt - Whether to keep the file extension.
    * @param {string} options.pattern - The date string pattern in the file name.
    * @param {boolean} options.alwaysIncludePattern - Whether to add date to the name of the first file.
    */
   constructor(filePath, options) {
     debug(`constructor: creating RollingFileWriteStream. path=${filePath}`);
     super(options);
     this.options = this._parseOption(options);
     this.fileObject = path.parse(filePath);
     if (this.fileObject.dir === "") {
       this.fileObject = path.parse(path.join(process.cwd(), filePath));
     }
     this.fileFormatter = fileNameFormatter({
       file: this.fileObject,
       alwaysIncludeDate: this.options.alwaysIncludePattern,
       needsIndex: this.options.maxSize < Number.MAX_SAFE_INTEGER,
       compress: this.options.compress,
       keepFileExt: this.options.keepFileExt
     });

     this.fileNameParser = fileNameParser({
       file: this.fileObject,
       keepFileExt: this.options.keepFileExt,
       pattern: this.options.pattern
     });

     this.state = {
       currentSize: 0
     };

     if (this.options.pattern) {
       this.state.currentDate = format(this.options.pattern, newNow());
     }

     this.filename = this.fileFormatter({
       index: 0,
       date: this.state.currentDate
     });
     if (["a", "a+", "as", "as+"].includes(this.options.flags)) {
       this._setExistingSizeAndDate();
     }

     debug(
       `constructor: create new file ${this.filename}, state=${JSON.stringify(
         this.state
       )}`
     );
     this._renewWriteStream();
   }

   _setExistingSizeAndDate() {
     try {
       const stats = fs.statSync(this.filename);
       this.state.currentSize = stats.size;
       if (this.options.pattern) {
         this.state.currentDate = format(this.options.pattern, stats.mtime);
       }
     } catch (e) {
       //file does not exist, that's fine - move along
       return;
     }
   }

   _parseOption(rawOptions) {
     const defaultOptions = {
       maxSize: Number.MAX_SAFE_INTEGER,
       numToKeep: Number.MAX_SAFE_INTEGER,
       encoding: "utf8",
       mode: parseInt("0644", 8),
       flags: "a",
       compress: false,
       keepFileExt: false,
       alwaysIncludePattern: false
     };
     const options = Object.assign({}, defaultOptions, rawOptions);
     if (options.maxSize <= 0) {
       throw new Error(`options.maxSize (${options.maxSize}) should be > 0`);
     }
     if (options.numToKeep <= 0) {
       throw new Error(`options.numToKeep (${options.numToKeep}) should be > 0`);
     }
     debug(
       `_parseOption: creating stream with option=${JSON.stringify(options)}`
     );
     return options;
   }

   _final(callback) {
     this.currentFileStream.end("", this.options.encoding, callback);
   }

   _write(chunk, encoding, callback) {
     this._shouldRoll().then(() => {
       debug(
         `_write: writing chunk. ` +
           `file=${this.currentFileStream.path} ` +
           `state=${JSON.stringify(this.state)} ` +
           `chunk=${chunk}`
       );
       this.currentFileStream.write(chunk, encoding, e => {
         this.state.currentSize += chunk.length;
         callback(e);
       });
     });
   }

   async _shouldRoll() {
     if (this._dateChanged() || this._tooBig()) {
       debug(
         `_shouldRoll: rolling because dateChanged? ${this._dateChanged()} or tooBig? ${this._tooBig()}`
       );
       await this._roll();
     }
   }

   _dateChanged() {
     return (
       this.state.currentDate &&
       this.state.currentDate !== format(this.options.pattern, newNow())
     );
   }

   _tooBig() {
     return this.state.currentSize >= this.options.maxSize;
   }

   _roll() {
     debug(`_roll: closing the current stream`);
     return new Promise((resolve, reject) => {
       this.currentFileStream.end("", this.options.encoding, () => {
         this._moveOldFiles()
           .then(resolve)
           .catch(reject);
       });
     });
   }

   async _moveOldFiles() {
     const files = await this._getExistingFiles();
     const todaysFiles = this.state.currentDate
       ? files.filter(f => f.date === this.state.currentDate)
       : files;
     for (let i = todaysFiles.length; i >= 0; i--) {
       debug(`_moveOldFiles: i = ${i}`);
       const sourceFilePath = this.fileFormatter({
         date: this.state.currentDate,
         index: i
       });
       const targetFilePath = this.fileFormatter({
         date: this.state.currentDate,
         index: i + 1
       });

       await moveAndMaybeCompressFile(
         sourceFilePath,
         targetFilePath,
         this.options.compress && i === 0
       );
     }

     this.state.currentSize = 0;
     this.state.currentDate = this.state.currentDate
       ? format(this.options.pattern, newNow())
       : null;
     debug(
       `_moveOldFiles: finished rolling files. state=${JSON.stringify(
         this.state
       )}`
     );
     this._renewWriteStream();
     // wait for the file to be open before cleaning up old ones,
     // otherwise the daysToKeep calculations can be off
     await new Promise((resolve, reject) => {
       this.currentFileStream.write("", "utf8", () => {
         this._clean()
           .then(resolve)
           .catch(reject);
       });
     });
   }

   // Sorted from the oldest to the latest
   async _getExistingFiles() {
     const files = await fs.readdir(this.fileObject.dir).catch(() => []);

     debug(`_getExistingFiles: files=${files}`);
     const existingFileDetails = files
       .map(n => this.fileNameParser(n))
       .filter(n => n);

     const getKey = n =>
       (n.timestamp ? n.timestamp : newNow().getTime()) - n.index;
     existingFileDetails.sort((a, b) => getKey(a) - getKey(b));

     return existingFileDetails;
   }

   _renewWriteStream() {
     fs.ensureDirSync(this.fileObject.dir);
     const filePath = this.fileFormatter({
       date: this.state.currentDate,
       index: 0
     });
     const ops = {
       flags: this.options.flags,
       encoding: this.options.encoding,
       mode: this.options.mode
     };
     this.currentFileStream = fs.createWriteStream(filePath, ops);
     this.currentFileStream.on("error", e => {
       this.emit("error", e);
     });
   }

   async _clean() {
     const existingFileDetails = await this._getExistingFiles();
     debug(
       `_clean: numToKeep = ${this.options.numToKeep}, existingFiles = ${existingFileDetails.length}`
     );
     debug("_clean: existing files are: ", existingFileDetails);
     if (this._tooManyFiles(existingFileDetails.length)) {
       const fileNamesToRemove = existingFileDetails
         .slice(0, existingFileDetails.length - this.options.numToKeep - 1)
         .map(f => path.format({ dir: this.fileObject.dir, base: f.filename }));
       await deleteFiles(fileNamesToRemove);
     }
   }

   _tooManyFiles(numFiles) {
     return this.options.numToKeep > 0 && numFiles > this.options.numToKeep;
   }
 }

 const deleteFiles = fileNames => {
   debug(`deleteFiles: files to delete: ${fileNames}`);
   return Promise.all(fileNames.map(f => fs.unlink(f).catch((e) => {
     debug(`deleteFiles: error when unlinking ${f}, ignoring. Error was ${e}`);
   })));
 };

 module.exports = RollingFileWriteStream;
	const debug = require("debug")("streamroller:RollingFileWriteStream");
	const fs = require("fs-extra");
	const path = require("path");
	const newNow = require("./now");
	const format = require("date-format");
	const { Writable } = require("stream");
	const fileNameFormatter = require("./fileNameFormatter");
	const fileNameParser = require("./fileNameParser");
	const moveAndMaybeCompressFile = require("./moveAndMaybeCompressFile");

	/**
	* RollingFileWriteStream is mainly used when writing to a file rolling by date or size.
	* RollingFileWriteStream inherits from stream.Writable
	*/
	class RollingFileWriteStream extends Writable {
	/**
	* Create a RollingFileWriteStream
	* @constructor
	* @param {string} filePath - The file path to write.
	* @param {object} options - The extra options
	* @param {number} options.numToKeep - The max numbers of files to keep.
	* @param {number} options.maxSize - The maxSize one file can reach. Unit is Byte.
	* This should be more than 1024. The default is Number.MAX_SAFE_INTEGER.
	* @param {string} options.mode - The mode of the files. The default is '0644'. Refer to stream.writable for more.
	* @param {string} options.flags - The default is 'a'. Refer to stream.flags for more.
	* @param {boolean} options.compress - Whether to compress backup files.
	* @param {boolean} options.keepFileExt - Whether to keep the file extension.
	* @param {string} options.pattern - The date string pattern in the file name.
	* @param {boolean} options.alwaysIncludePattern - Whether to add date to the name of the first file.
	*/
	constructor(filePath, options) {
	debug(`constructor: creating RollingFileWriteStream. path=${filePath}`);
	super(options);
	this.options = this._parseOption(options);
	this.fileObject = path.parse(filePath);
	if (this.fileObject.dir === "") {
	this.fileObject = path.parse(path.join(process.cwd(), filePath));
	}
	this.fileFormatter = fileNameFormatter({
	file: this.fileObject,
	alwaysIncludeDate: this.options.alwaysIncludePattern,
	needsIndex: this.options.maxSize < Number.MAX_SAFE_INTEGER,
	compress: this.options.compress,
	keepFileExt: this.options.keepFileExt
	});

	this.fileNameParser = fileNameParser({
	file: this.fileObject,
	keepFileExt: this.options.keepFileExt,
	pattern: this.options.pattern
	});

	this.state = {
	currentSize: 0
	};

	if (this.options.pattern) {
	this.state.currentDate = format(this.options.pattern, newNow());
	}

	this.filename = this.fileFormatter({
	index: 0,
	date: this.state.currentDate
	});
	if (["a", "a+", "as", "as+"].includes(this.options.flags)) {
	this._setExistingSizeAndDate();
	}

	debug(
	`constructor: create new file ${this.filename}, state=${JSON.stringify(
	this.state
	)}`
	);
	this._renewWriteStream();
	}

	_setExistingSizeAndDate() {
	try {
	const stats = fs.statSync(this.filename);
	this.state.currentSize = stats.size;
	if (this.options.pattern) {
	this.state.currentDate = format(this.options.pattern, stats.mtime);
	}
	} catch (e) {
	//file does not exist, that's fine - move along
	return;
	}
	}

	_parseOption(rawOptions) {
	const defaultOptions = {
	maxSize: Number.MAX_SAFE_INTEGER,
	numToKeep: Number.MAX_SAFE_INTEGER,
	encoding: "utf8",
	mode: parseInt("0644", 8),
	flags: "a",
	compress: false,
	keepFileExt: false,
	alwaysIncludePattern: false
	};
	const options = Object.assign({}, defaultOptions, rawOptions);
	if (options.maxSize <= 0) {
	throw new Error(`options.maxSize (${options.maxSize}) should be > 0`);
	}
	if (options.numToKeep <= 0) {
	throw new Error(`options.numToKeep (${options.numToKeep}) should be > 0`);
	}
	debug(
	`_parseOption: creating stream with option=${JSON.stringify(options)}`
	);
	return options;
	}

	_final(callback) {
	this.currentFileStream.end("", this.options.encoding, callback);
	}

	_write(chunk, encoding, callback) {
	this._shouldRoll().then(() => {
	debug(
	`_write: writing chunk. ` +
	`file=${this.currentFileStream.path} ` +
	`state=${JSON.stringify(this.state)} ` +
	`chunk=${chunk}`
	);
	this.currentFileStream.write(chunk, encoding, e => {
	this.state.currentSize += chunk.length;
	callback(e);
	});
	});
	}

	async _shouldRoll() {
	if (this._dateChanged() \|\| this._tooBig()) {
	debug(
	`_shouldRoll: rolling because dateChanged? ${this._dateChanged()} or tooBig? ${this._tooBig()}`
	);
	await this._roll();
	}
	}

	_dateChanged() {
	return (
	this.state.currentDate &&
	this.state.currentDate !== format(this.options.pattern, newNow())
	);
	}

	_tooBig() {
	return this.state.currentSize >= this.options.maxSize;
	}

	_roll() {
	debug(`_roll: closing the current stream`);
	return new Promise((resolve, reject) => {
	this.currentFileStream.end("", this.options.encoding, () => {
	this._moveOldFiles()
	.then(resolve)
	.catch(reject);
	});
	});
	}

	async _moveOldFiles() {
	const files = await this._getExistingFiles();
	const todaysFiles = this.state.currentDate
	? files.filter(f => f.date === this.state.currentDate)
	: files;
	for (let i = todaysFiles.length; i >= 0; i--) {
	debug(`_moveOldFiles: i = ${i}`);
	const sourceFilePath = this.fileFormatter({
	date: this.state.currentDate,
	index: i
	});
	const targetFilePath = this.fileFormatter({
	date: this.state.currentDate,
	index: i + 1
	});

	await moveAndMaybeCompressFile(
	sourceFilePath,
	targetFilePath,
	this.options.compress && i === 0
	);
	}

	this.state.currentSize = 0;
	this.state.currentDate = this.state.currentDate
	? format(this.options.pattern, newNow())
	: null;
	debug(
	`_moveOldFiles: finished rolling files. state=${JSON.stringify(
	this.state
	)}`
	);
	this._renewWriteStream();
	// wait for the file to be open before cleaning up old ones,
	// otherwise the daysToKeep calculations can be off
	await new Promise((resolve, reject) => {
	this.currentFileStream.write("", "utf8", () => {
	this._clean()
	.then(resolve)
	.catch(reject);
	});
	});
	}

	// Sorted from the oldest to the latest
	async _getExistingFiles() {
	const files = await fs.readdir(this.fileObject.dir).catch(() => []);

	debug(`_getExistingFiles: files=${files}`);
	const existingFileDetails = files
	.map(n => this.fileNameParser(n))
	.filter(n => n);

	const getKey = n =>
	(n.timestamp ? n.timestamp : newNow().getTime()) - n.index;
	existingFileDetails.sort((a, b) => getKey(a) - getKey(b));

	return existingFileDetails;
	}

	_renewWriteStream() {
	fs.ensureDirSync(this.fileObject.dir);
	const filePath = this.fileFormatter({
	date: this.state.currentDate,
	index: 0
	});
	const ops = {
	flags: this.options.flags,
	encoding: this.options.encoding,
	mode: this.options.mode
	};
	this.currentFileStream = fs.createWriteStream(filePath, ops);
	this.currentFileStream.on("error", e => {
	this.emit("error", e);
	});
	}

	async _clean() {
	const existingFileDetails = await this._getExistingFiles();
	debug(
	`_clean: numToKeep = ${this.options.numToKeep}, existingFiles = ${existingFileDetails.length}`
	);
	debug("_clean: existing files are: ", existingFileDetails);
	if (this._tooManyFiles(existingFileDetails.length)) {
	const fileNamesToRemove = existingFileDetails
	.slice(0, existingFileDetails.length - this.options.numToKeep - 1)
	.map(f => path.format({ dir: this.fileObject.dir, base: f.filename }));
	await deleteFiles(fileNamesToRemove);
	}
	}

	_tooManyFiles(numFiles) {
	return this.options.numToKeep > 0 && numFiles > this.options.numToKeep;
	}
	}

	const deleteFiles = fileNames => {
	debug(`deleteFiles: files to delete: ${fileNames}`);
	return Promise.all(fileNames.map(f => fs.unlink(f).catch((e) => {
	debug(`deleteFiles: error when unlinking ${f}, ignoring. Error was ${e}`);
	})));
	};

	module.exports = RollingFileWriteStream;