[6a3a178] | 1 | const debug = require("debug")("streamroller:RollingFileWriteStream");
|
---|
| 2 | const fs = require("fs-extra");
|
---|
| 3 | const path = require("path");
|
---|
| 4 | const newNow = require("./now");
|
---|
| 5 | const format = require("date-format");
|
---|
| 6 | const { Writable } = require("stream");
|
---|
| 7 | const fileNameFormatter = require("./fileNameFormatter");
|
---|
| 8 | const fileNameParser = require("./fileNameParser");
|
---|
| 9 | const moveAndMaybeCompressFile = require("./moveAndMaybeCompressFile");
|
---|
| 10 |
|
---|
| 11 | /**
|
---|
| 12 | * RollingFileWriteStream is mainly used when writing to a file rolling by date or size.
|
---|
| 13 | * RollingFileWriteStream inherits from stream.Writable
|
---|
| 14 | */
|
---|
| 15 | class RollingFileWriteStream extends Writable {
|
---|
| 16 | /**
|
---|
| 17 | * Create a RollingFileWriteStream
|
---|
| 18 | * @constructor
|
---|
| 19 | * @param {string} filePath - The file path to write.
|
---|
| 20 | * @param {object} options - The extra options
|
---|
| 21 | * @param {number} options.numToKeep - The max numbers of files to keep.
|
---|
| 22 | * @param {number} options.maxSize - The maxSize one file can reach. Unit is Byte.
|
---|
| 23 | * This should be more than 1024. The default is Number.MAX_SAFE_INTEGER.
|
---|
| 24 | * @param {string} options.mode - The mode of the files. The default is '0644'. Refer to stream.writable for more.
|
---|
| 25 | * @param {string} options.flags - The default is 'a'. Refer to stream.flags for more.
|
---|
| 26 | * @param {boolean} options.compress - Whether to compress backup files.
|
---|
| 27 | * @param {boolean} options.keepFileExt - Whether to keep the file extension.
|
---|
| 28 | * @param {string} options.pattern - The date string pattern in the file name.
|
---|
| 29 | * @param {boolean} options.alwaysIncludePattern - Whether to add date to the name of the first file.
|
---|
| 30 | */
|
---|
| 31 | constructor(filePath, options) {
|
---|
| 32 | debug(`constructor: creating RollingFileWriteStream. path=${filePath}`);
|
---|
| 33 | super(options);
|
---|
| 34 | this.options = this._parseOption(options);
|
---|
| 35 | this.fileObject = path.parse(filePath);
|
---|
| 36 | if (this.fileObject.dir === "") {
|
---|
| 37 | this.fileObject = path.parse(path.join(process.cwd(), filePath));
|
---|
| 38 | }
|
---|
| 39 | this.fileFormatter = fileNameFormatter({
|
---|
| 40 | file: this.fileObject,
|
---|
| 41 | alwaysIncludeDate: this.options.alwaysIncludePattern,
|
---|
| 42 | needsIndex: this.options.maxSize < Number.MAX_SAFE_INTEGER,
|
---|
| 43 | compress: this.options.compress,
|
---|
| 44 | keepFileExt: this.options.keepFileExt
|
---|
| 45 | });
|
---|
| 46 |
|
---|
| 47 | this.fileNameParser = fileNameParser({
|
---|
| 48 | file: this.fileObject,
|
---|
| 49 | keepFileExt: this.options.keepFileExt,
|
---|
| 50 | pattern: this.options.pattern
|
---|
| 51 | });
|
---|
| 52 |
|
---|
| 53 | this.state = {
|
---|
| 54 | currentSize: 0
|
---|
| 55 | };
|
---|
| 56 |
|
---|
| 57 | if (this.options.pattern) {
|
---|
| 58 | this.state.currentDate = format(this.options.pattern, newNow());
|
---|
| 59 | }
|
---|
| 60 |
|
---|
| 61 | this.filename = this.fileFormatter({
|
---|
| 62 | index: 0,
|
---|
| 63 | date: this.state.currentDate
|
---|
| 64 | });
|
---|
| 65 | if (["a", "a+", "as", "as+"].includes(this.options.flags)) {
|
---|
| 66 | this._setExistingSizeAndDate();
|
---|
| 67 | }
|
---|
| 68 |
|
---|
| 69 | debug(
|
---|
| 70 | `constructor: create new file ${this.filename}, state=${JSON.stringify(
|
---|
| 71 | this.state
|
---|
| 72 | )}`
|
---|
| 73 | );
|
---|
| 74 | this._renewWriteStream();
|
---|
| 75 | }
|
---|
| 76 |
|
---|
| 77 | _setExistingSizeAndDate() {
|
---|
| 78 | try {
|
---|
| 79 | const stats = fs.statSync(this.filename);
|
---|
| 80 | this.state.currentSize = stats.size;
|
---|
| 81 | if (this.options.pattern) {
|
---|
| 82 | this.state.currentDate = format(this.options.pattern, stats.mtime);
|
---|
| 83 | }
|
---|
| 84 | } catch (e) {
|
---|
| 85 | //file does not exist, that's fine - move along
|
---|
| 86 | return;
|
---|
| 87 | }
|
---|
| 88 | }
|
---|
| 89 |
|
---|
| 90 | _parseOption(rawOptions) {
|
---|
| 91 | const defaultOptions = {
|
---|
| 92 | maxSize: Number.MAX_SAFE_INTEGER,
|
---|
| 93 | numToKeep: Number.MAX_SAFE_INTEGER,
|
---|
| 94 | encoding: "utf8",
|
---|
| 95 | mode: parseInt("0644", 8),
|
---|
| 96 | flags: "a",
|
---|
| 97 | compress: false,
|
---|
| 98 | keepFileExt: false,
|
---|
| 99 | alwaysIncludePattern: false
|
---|
| 100 | };
|
---|
| 101 | const options = Object.assign({}, defaultOptions, rawOptions);
|
---|
| 102 | if (options.maxSize <= 0) {
|
---|
| 103 | throw new Error(`options.maxSize (${options.maxSize}) should be > 0`);
|
---|
| 104 | }
|
---|
| 105 | if (options.numToKeep <= 0) {
|
---|
| 106 | throw new Error(`options.numToKeep (${options.numToKeep}) should be > 0`);
|
---|
| 107 | }
|
---|
| 108 | debug(
|
---|
| 109 | `_parseOption: creating stream with option=${JSON.stringify(options)}`
|
---|
| 110 | );
|
---|
| 111 | return options;
|
---|
| 112 | }
|
---|
| 113 |
|
---|
| 114 | _final(callback) {
|
---|
| 115 | this.currentFileStream.end("", this.options.encoding, callback);
|
---|
| 116 | }
|
---|
| 117 |
|
---|
| 118 | _write(chunk, encoding, callback) {
|
---|
| 119 | this._shouldRoll().then(() => {
|
---|
| 120 | debug(
|
---|
| 121 | `_write: writing chunk. ` +
|
---|
| 122 | `file=${this.currentFileStream.path} ` +
|
---|
| 123 | `state=${JSON.stringify(this.state)} ` +
|
---|
| 124 | `chunk=${chunk}`
|
---|
| 125 | );
|
---|
| 126 | this.currentFileStream.write(chunk, encoding, e => {
|
---|
| 127 | this.state.currentSize += chunk.length;
|
---|
| 128 | callback(e);
|
---|
| 129 | });
|
---|
| 130 | });
|
---|
| 131 | }
|
---|
| 132 |
|
---|
| 133 | async _shouldRoll() {
|
---|
| 134 | if (this._dateChanged() || this._tooBig()) {
|
---|
| 135 | debug(
|
---|
| 136 | `_shouldRoll: rolling because dateChanged? ${this._dateChanged()} or tooBig? ${this._tooBig()}`
|
---|
| 137 | );
|
---|
| 138 | await this._roll();
|
---|
| 139 | }
|
---|
| 140 | }
|
---|
| 141 |
|
---|
| 142 | _dateChanged() {
|
---|
| 143 | return (
|
---|
| 144 | this.state.currentDate &&
|
---|
| 145 | this.state.currentDate !== format(this.options.pattern, newNow())
|
---|
| 146 | );
|
---|
| 147 | }
|
---|
| 148 |
|
---|
| 149 | _tooBig() {
|
---|
| 150 | return this.state.currentSize >= this.options.maxSize;
|
---|
| 151 | }
|
---|
| 152 |
|
---|
| 153 | _roll() {
|
---|
| 154 | debug(`_roll: closing the current stream`);
|
---|
| 155 | return new Promise((resolve, reject) => {
|
---|
| 156 | this.currentFileStream.end("", this.options.encoding, () => {
|
---|
| 157 | this._moveOldFiles()
|
---|
| 158 | .then(resolve)
|
---|
| 159 | .catch(reject);
|
---|
| 160 | });
|
---|
| 161 | });
|
---|
| 162 | }
|
---|
| 163 |
|
---|
| 164 | async _moveOldFiles() {
|
---|
| 165 | const files = await this._getExistingFiles();
|
---|
| 166 | const todaysFiles = this.state.currentDate
|
---|
| 167 | ? files.filter(f => f.date === this.state.currentDate)
|
---|
| 168 | : files;
|
---|
| 169 | for (let i = todaysFiles.length; i >= 0; i--) {
|
---|
| 170 | debug(`_moveOldFiles: i = ${i}`);
|
---|
| 171 | const sourceFilePath = this.fileFormatter({
|
---|
| 172 | date: this.state.currentDate,
|
---|
| 173 | index: i
|
---|
| 174 | });
|
---|
| 175 | const targetFilePath = this.fileFormatter({
|
---|
| 176 | date: this.state.currentDate,
|
---|
| 177 | index: i + 1
|
---|
| 178 | });
|
---|
| 179 |
|
---|
| 180 | await moveAndMaybeCompressFile(
|
---|
| 181 | sourceFilePath,
|
---|
| 182 | targetFilePath,
|
---|
| 183 | this.options.compress && i === 0
|
---|
| 184 | );
|
---|
| 185 | }
|
---|
| 186 |
|
---|
| 187 | this.state.currentSize = 0;
|
---|
| 188 | this.state.currentDate = this.state.currentDate
|
---|
| 189 | ? format(this.options.pattern, newNow())
|
---|
| 190 | : null;
|
---|
| 191 | debug(
|
---|
| 192 | `_moveOldFiles: finished rolling files. state=${JSON.stringify(
|
---|
| 193 | this.state
|
---|
| 194 | )}`
|
---|
| 195 | );
|
---|
| 196 | this._renewWriteStream();
|
---|
| 197 | // wait for the file to be open before cleaning up old ones,
|
---|
| 198 | // otherwise the daysToKeep calculations can be off
|
---|
| 199 | await new Promise((resolve, reject) => {
|
---|
| 200 | this.currentFileStream.write("", "utf8", () => {
|
---|
| 201 | this._clean()
|
---|
| 202 | .then(resolve)
|
---|
| 203 | .catch(reject);
|
---|
| 204 | });
|
---|
| 205 | });
|
---|
| 206 | }
|
---|
| 207 |
|
---|
| 208 | // Sorted from the oldest to the latest
|
---|
| 209 | async _getExistingFiles() {
|
---|
| 210 | const files = await fs.readdir(this.fileObject.dir).catch(() => []);
|
---|
| 211 |
|
---|
| 212 | debug(`_getExistingFiles: files=${files}`);
|
---|
| 213 | const existingFileDetails = files
|
---|
| 214 | .map(n => this.fileNameParser(n))
|
---|
| 215 | .filter(n => n);
|
---|
| 216 |
|
---|
| 217 | const getKey = n =>
|
---|
| 218 | (n.timestamp ? n.timestamp : newNow().getTime()) - n.index;
|
---|
| 219 | existingFileDetails.sort((a, b) => getKey(a) - getKey(b));
|
---|
| 220 |
|
---|
| 221 | return existingFileDetails;
|
---|
| 222 | }
|
---|
| 223 |
|
---|
| 224 | _renewWriteStream() {
|
---|
| 225 | fs.ensureDirSync(this.fileObject.dir);
|
---|
| 226 | const filePath = this.fileFormatter({
|
---|
| 227 | date: this.state.currentDate,
|
---|
| 228 | index: 0
|
---|
| 229 | });
|
---|
| 230 | const ops = {
|
---|
| 231 | flags: this.options.flags,
|
---|
| 232 | encoding: this.options.encoding,
|
---|
| 233 | mode: this.options.mode
|
---|
| 234 | };
|
---|
| 235 | this.currentFileStream = fs.createWriteStream(filePath, ops);
|
---|
| 236 | this.currentFileStream.on("error", e => {
|
---|
| 237 | this.emit("error", e);
|
---|
| 238 | });
|
---|
| 239 | }
|
---|
| 240 |
|
---|
| 241 | async _clean() {
|
---|
| 242 | const existingFileDetails = await this._getExistingFiles();
|
---|
| 243 | debug(
|
---|
| 244 | `_clean: numToKeep = ${this.options.numToKeep}, existingFiles = ${existingFileDetails.length}`
|
---|
| 245 | );
|
---|
| 246 | debug("_clean: existing files are: ", existingFileDetails);
|
---|
| 247 | if (this._tooManyFiles(existingFileDetails.length)) {
|
---|
| 248 | const fileNamesToRemove = existingFileDetails
|
---|
| 249 | .slice(0, existingFileDetails.length - this.options.numToKeep - 1)
|
---|
| 250 | .map(f => path.format({ dir: this.fileObject.dir, base: f.filename }));
|
---|
| 251 | await deleteFiles(fileNamesToRemove);
|
---|
| 252 | }
|
---|
| 253 | }
|
---|
| 254 |
|
---|
| 255 | _tooManyFiles(numFiles) {
|
---|
| 256 | return this.options.numToKeep > 0 && numFiles > this.options.numToKeep;
|
---|
| 257 | }
|
---|
| 258 | }
|
---|
| 259 |
|
---|
| 260 | const deleteFiles = fileNames => {
|
---|
| 261 | debug(`deleteFiles: files to delete: ${fileNames}`);
|
---|
| 262 | return Promise.all(fileNames.map(f => fs.unlink(f).catch((e) => {
|
---|
| 263 | debug(`deleteFiles: error when unlinking ${f}, ignoring. Error was ${e}`);
|
---|
| 264 | })));
|
---|
| 265 | };
|
---|
| 266 |
|
---|
| 267 | module.exports = RollingFileWriteStream;
|
---|