import fs from 'fs' ; import JSZip from 'jszip' ; import mkdirp from 'mkdirp' ; import path from 'path' ; import { MsgWriter , TimestampMsg } from './msg_writer' ; export class RotatedFileWriter implements MsgWriter { private rootDir : string ; private filenamePrefix : string ; private interval : 'Minutely' | 'Hourly' | 'Daily' ; private timestamp : number ; private fileStream : fs . WriteStream ; constructor ( rootDir , filenamePrefix = '' , interval = 'Hourly' ) { this . rootDir = rootDir ; this . filenamePrefix = filenamePrefix ; if ( ! fs . existsSync ( rootDir ) ) { mkdirp . sync ( rootDir ) ; } this . interval = interval ; const intervalMilliseconds = RotatedFileWriter . getIntervalMilliseconds ( this . interval ) ; this . timestamp = Math . floor ( Date . now ( ) / intervalMilliseconds ) * intervalMilliseconds ; this . fileStream = this . createWriteStream ( ) ; } public async write ( msg ) { const intervalMilliseconds = RotatedFileWriter . getIntervalMilliseconds ( this . interval ) ; const nextTimestamp = this . timestamp + intervalMilliseconds ; if ( msg . timestamp >= nextTimestamp ) { const fileIn = this . fileStream . path as string ; this . fileStream . end ( ( ) => { RotatedFileWriter . compress ( fileIn , fileIn . replace ( '.json' , '.zip' ) ) ; } ) ; this . timestamp = nextTimestamp ; this . fileStream = this . createWriteStream ( ) ; } this . fileStream . write ( `${ JSON . stringify ( msg ) }\n` ) ; } private static getIntervalMilliseconds ( interval = 'Hourly' ) { switch ( interval ) { case 'Minutely' : return 60 * 1000 ; case 'Hourly' : return 3600 * 1000 ; case 'Daily' : return 24 * 3600 * 1000 ; default : throw new Error ( `Unknown interval ${ interval }` ) ; } } private createWriteStream ( ) { let filename = '' ; switch ( this . interval ) { case 'Minutely' : filename = `${ this . filenamePrefix }${ new Date ( this . timestamp ) . toISOString ( ) . slice ( 0 , 16 ) }.json` ; break ; case 'Hourly' : filename = `${ this . filenamePrefix }${ new Date ( this . timestamp ) . toISOString ( ) . slice ( 0 , 13 ) }.json` ; break ; case 'Daily' : filename = `${ this . filenamePrefix }${ new Date ( this . timestamp ) . toISOString ( ) . slice ( 0 , 10 ) }.json` ; break ; default : throw new Error ( `Unknown interval ${ this . interval }` ) ; } const filePath = path . join ( this . rootDir , filename ) ; return fs . createWriteStream ( filePath , { flags : 'a' , encoding : 'utf8' , } ) ; } private static async compress ( fileIn , fileOut ) { const zip = new JSZip ( ) ; zip . file ( path . parse ( fileIn ) . base , fs . createReadStream ( fileIn , 'utf8' ) ) ; zip . generateNodeStream ( { type : 'nodebuffer' , compression : 'DEFLATE' , compressionOptions : { level : 9 , } , streamFiles : true , } ) . pipe ( fs . createWriteStream ( fileOut ) ) . on ( 'finish' , ( ) => { fs . unlinkSync ( fileIn ) ; } ) ; } }