import fs from 'fs' ; import { LogFileOperationResult , LogFileParsingState , ParsingMetadata , StatefulLogEvent , } from 'root/app/log-parser/model' ; import { parseEvent } from 'root/app/log-parser/parsing' ; import { locateMostRecentDate } from 'root/app/mtga_dir_ops' ; function shouldStopParsing ( allEvents , options ) { if ( allEvents . length === 0 ) { return false ; } if ( allEvents . length > options . logParser . batchSize ) { return true ; } const lastEvent = allEvents [ allEvents . length - 1 ] ; return lastEvent . name === options . userChangeEvent ; } export async function getEvents ( path , state , options , oldlog ? ) { return new Promise < LogFileOperationResult < StatefulLogEvent [ ] > > ( ( resolve , reject ) => { const fileStream = fs . createReadStream ( path , { encoding : 'utf8' , autoClose : true , start : state . bytesRead , } ) ; let bytesRead = state . bytesRead ; let currentEvent ; const allEvents = [ ] ; if ( ! oldlog && ( state . timestamp === undefined || state . timestamp === 1 || state . timestamp < 0 ) ) { state . timestamp = locateMostRecentDate ( ) . date ; } fileStream . on ( 'data' , ( chunk ) => { bytesRead += chunk . length ; let chunkCursor = 0 ; if ( currentEvent !== undefined ) { const res = chunk . split ( /\r?\n/ , 2 ) ; if ( res . length === 1 ) { currentEvent += chunk ; return ; } else { currentEvent += res [ 0 ] ; chunkCursor += res [ 0 ] . length ; parseEvent ( currentEvent , state , options , oldlog ) . forEach ( ( e ) => allEvents . push ( e ) ) ; if ( shouldStopParsing ( allEvents , options ) ) { fileStream . close ( ) ; resolve ( [ allEvents , { ... state , bytesRead : bytesRead - res [ 1 ] . length } ] ) ; return ; } currentEvent = undefined ; } } while ( true ) { const prefIndex = chunk . indexOf ( options . eventPrefix , chunkCursor ) ; const prefIndexExtra = chunk . indexOf ( options . eventPrefixExtra , chunkCursor ) ; const nextPrefixIndex = prefIndex < prefIndexExtra ? prefIndex !== - 1 ? prefIndex : prefIndexExtra : prefIndexExtra !== - 1 ? prefIndexExtra : prefIndex ; const eventPrefix = prefIndex < prefIndexExtra ? prefIndex !== - 1 ? options . eventPrefix : options . eventPrefixExtra : prefIndexExtra !== - 1 ? options . eventPrefixExtra : options . eventPrefix ; if ( nextPrefixIndex === - 1 ) { return ; } let nextLineBreakIndex = chunk . indexOf ( '\n' , nextPrefixIndex + eventPrefix . length ) ; if ( nextLineBreakIndex === - 1 ) { currentEvent = chunk . slice ( nextPrefixIndex + eventPrefix . length , chunk . length ) ; return ; } if ( chunk [ nextLineBreakIndex + 1 ] === '{' || chunk [ nextLineBreakIndex + 1 ] === '<' ) { let lineBreakAfter = chunk . indexOf ( '\n' , nextLineBreakIndex + 1 ) ; if ( chunk [ nextLineBreakIndex + 1 ] === '<' ) { lineBreakAfter = chunk . indexOf ( '\n' , lineBreakAfter + 1 ) ; } if ( lineBreakAfter === - 1 ) { currentEvent = chunk . slice ( nextPrefixIndex + eventPrefix . length , chunk . length ) . split ( /\r?\n/ ) . join ( chunk [ nextLineBreakIndex + 1 ] === '<' ? ' ' : '' ) ; return ; } if ( chunk [ lineBreakAfter - 1 ] === '\r' ) { lineBreakAfter -- ; } const eventString = chunk . slice ( nextPrefixIndex + eventPrefix . length , lineBreakAfter ) . split ( /\r?\n/ ) . join ( chunk [ nextLineBreakIndex + 1 ] === '<' ? ' ' : '' ) ; parseEvent ( eventString , state , options ) . forEach ( ( e ) => allEvents . push ( e ) ) ; if ( shouldStopParsing ( allEvents , options ) ) { fileStream . close ( ) ; resolve ( [ allEvents , { ... state , bytesRead : bytesRead - chunk . length + lineBreakAfter } ] ) ; return ; } chunkCursor = lineBreakAfter ; } else { if ( chunk [ nextLineBreakIndex - 1 ] === '\r' ) { nextLineBreakIndex -- ; } const eventString = chunk . slice ( nextPrefixIndex + eventPrefix . length , nextLineBreakIndex ) ; parseEvent ( eventString , state , options ) . forEach ( ( e ) => allEvents . push ( e ) ) ; if ( shouldStopParsing ( allEvents , options ) ) { fileStream . close ( ) ; resolve ( [ allEvents , { ... state , bytesRead : bytesRead - chunk . length + nextLineBreakIndex } ] ) ; return ; } chunkCursor = nextLineBreakIndex ; } } } ) ; fileStream . on ( 'end' , ( ) => { fileStream . close ( ) ; resolve ( [ allEvents , { ... state , bytesRead } ] ) ; } ) ; fileStream . on ( 'error' , ( err ) => { reject ( err ) ; fileStream . close ( ) ; } ) ; } ) ; }