import { AFileParser , IMapItem } from './interface' ; import * as path from 'path' ; import ft from 'file-type' ; import * as fs from 'fs' ; import * as Tools from '../../tools/index' ; import indexer , { CancelablePromise , Processor , Progress } from 'indexer-neon' ; import ServiceNotifications , { ENotificationType } from '../../services/service.notifications' ; import ServiceStreams from '../../services/service.streams' ; import Logger from '../../tools/env.logger' ; export default class FileParser extends AFileParser { private _guid : string | undefined ; private _logger : Logger = new Logger ( 'Plain Text Indexing' ) ; private _task : CancelablePromise < void , void , Processor . TIndexAsyncEvents , Processor . TIndexAsyncEventObject > | undefined ; constructor ( ) { super ( ) ; } public static getAlias ( ) { return 'text' ; } public destroy ( ) { return this . abort ( ) ; } public getName ( ) { return 'text format' ; } public getAlias ( ) { return FileParser . getAlias ( ) ; } public getMeta ( ) { return 'plain/text' ; } public getExtnameFilters ( ) { return [ ] ; } public getExtensions ( ) { return [ ] ; } public isSupported ( file ) { return new Promise ( ( resolve , reject ) => { fs . open ( file , 'r' , ( openError , fd ) => { if ( openError ) { return reject ( openError ) ; } const buffer = Buffer . alloc ( ft . minimumBytes ) ; fs . read ( fd , buffer , 0 , ft . minimumBytes , 0 , ( readError , read , buf ) => { if ( readError ) { return reject ( readError ) ; } const type = ft ( buf ) ; if ( type === undefined ) { resolve ( false ) ; } else if ( type . mime . indexOf ( 'text' ) !== - 1 || type . mime . indexOf ( 'application' ) !== - 1 ) { resolve ( true ) ; } resolve ( false ) ; } ) ; } ) ; } ) ; } public parseAndIndex ( srcFile , destFile , sourceId , options , onMapUpdated ? , onProgress ? ) { return new Tools . CancelablePromise < IMapItem [ ] , void > ( ( resolve , reject , cancel , self ) => { this . _guid = ServiceStreams . getActiveStreamId ( ) ; let completeTicks = 0 ; const collectedChunks = [ ] ; const hrstart = process . hrtime ( ) ; this . _task = indexer . indexAsync ( srcFile , destFile , sourceId . toString ( ) , { chunkSize : 500 } ) . then ( ( ) => { if ( onProgress !== undefined ) { onProgress ( { ellapsed : completeTicks , total : completeTicks , } ) ; } resolve ( collectedChunks ) ; } ) . catch ( ( error ) => { ServiceNotifications . notify ( { message : error . message . length > 1500 ? `${ error . message . substr ( 0 , 1500 ) }...` : error . message , caption : `Error with: ${ path . basename ( srcFile ) }` , session : this . _guid , file : sourceId . toString ( ) , type : ENotificationType . error , } ) ; reject ( error ) ; } ) . canceled ( ( ) => { cancel ( ) ; } ) . finally ( ( ) => { this . _task = undefined ; const hrend = process . hrtime ( hrstart ) ; const ms = Math . round ( hrend [ 0 ] * 1000 + hrend [ 1 ] / 1000000 ) ; this . _logger . debug ( 'parseAndIndex task finished' ) ; this . _logger . debug ( 'Execution time for indexing : ' + ms + 'ms' ) ; } ) . on ( 'chunk' , ( event ) => { if ( onMapUpdated !== undefined ) { const mapItem = { rows : { from : event . rowsStart , to : event . rowsEnd } , bytes : { from : event . bytesStart , to : event . bytesEnd } , } ; onMapUpdated ( [ mapItem ] ) ; collectedChunks . push ( mapItem ) ; } } ) . on ( 'progress' , ( event ) => { if ( onProgress !== undefined ) { completeTicks = event . total ; onProgress ( event ) ; } } ) . on ( 'notification' , ( event ) => { ServiceNotifications . notifyFromNeon ( event , 'Indexing' , this . _guid , srcFile ) ; } ) ; } ) ; } public abort ( ) { return new Promise ( ( resolve ) => { if ( this . _task === undefined ) { return resolve ( ) ; } this . _task . canceled ( ( ) => { resolve ( ) ; } ) . abort ( ) ; } ) ; } }