import Logger from '../../../tools/env.logger' ; import guid from '../../../tools/tools.guid' ; import State from '../state' ; import ServiceStreams from '../../../services/service.streams' ; import ControllerStreamProcessor from '../../stream.main/controller' ; import * as fs from 'fs' ; import { IMapItem } from '../file.map' ; import { CancelablePromise } from '../../../tools/promise.cancelable' ; import { EventEmitter } from 'events' ; import { OperationSearch , IMapChunkEvent } from './operation.search' ; import { OperationAppend } from './operation.append' ; import { OperationInspecting , IScaledMapData , IIndexAround } from './operation.inspecting' ; export { IMapChunkEvent } ; export interface IRange { from : number ; to : number ; } export class SearchEngine extends EventEmitter { public static Events = { onMapUpdated : 'onMapUpdated' , } ; private _logger : Logger ; private _state : State ; private _stream : ControllerStreamProcessor ; private _stock : { search : Map < string , CancelablePromise < any , void > > ; inspecting : Map < string , CancelablePromise < any , void > > ; } = { search : new Map ( ) , inspecting : new Map ( ) , } ; private _operations : { search : OperationSearch ; append : OperationAppend ; inspecting : OperationInspecting ; } ; private _size : number = 0 ; constructor ( state , stream ) { super ( ) ; this . _state = state ; this . _stream = stream ; this . _logger = new Logger ( `ControllerSearchEngine: ${ this . _state . getGuid ( ) }` ) ; this . _operations = { search : new OperationSearch ( this . _state . getGuid ( ) , this . _state . getStreamFile ( ) , this . _state . getSearchFile ( ) ) , append : new OperationAppend ( this . _state . getGuid ( ) , this . _state . getStreamFile ( ) , this . _state . getSearchFile ( ) ) , inspecting : new OperationInspecting ( this . _state . getGuid ( ) , this . _state . getStreamFile ( ) , this . _state . getSearchFile ( ) ) , } ; this . _onMapUpdated = this . _onMapUpdated . bind ( this ) ; this . _operations . search . on ( OperationSearch . Events . onMapUpdated , this . _onMapUpdated ) ; this . _operations . append . on ( OperationAppend . Events . onMapUpdated , this . _onMapUpdated ) ; } public destroy ( ) { return new Promise ( ( resolve , reject ) => { this . drop ( ) . catch ( ( err ) => { this . _logger . error ( `Unexpected error: ${ err . message }` ) ; } ) . finally ( ( ) => { this . removeAllListeners ( ) ; this . cancel ( ) ; resolve ( ) ; } ) ; } ) ; } public drop ( ) { return new Promise ( ( resolve ) => { resolve ( ) ; } ) ; } public search ( requests , to ? ) { const taskId = guid ( ) ; if ( this . _stock . search . size !== 0 ) { return new Error ( `Fail to start search because previous wasn't finished.` ) ; } let error ; const isAppending = typeof to === 'number' ; if ( ! isAppending ) { this . _operations . append . drop ( ) ; this . _operations . search . drop ( ) ; this . _operations . inspecting . drop ( ) ; } if ( typeof to === 'number' ) { error = this . _operations . append . perform ( requests , to , taskId ) ; } else { error = this . _operations . search . perform ( requests , taskId ) ; } if ( error instanceof Error ) { this . _logger . error ( `Fail perform operation due error: ${ error . message }` ) ; return error ; } const task = ( error as CancelablePromise < IMapItem [ ] , void > ) ; return new CancelablePromise < IMapItem [ ] , void > ( ( resolve , reject ) => { this . _stock . search . set ( taskId , task ) ; this . _stream . addProgressSession ( taskId , 'search' ) ; this . _stream . updateProgressSession ( taskId , 0 ) ; task . then ( ( map ) => { fs . stat ( this . _state . getSearchFile ( ) , ( err , stats ) => { if ( ! err ) { this . _size = stats . size ; } resolve ( map ) ; } ) ; } ) . cancel ( ( ) => { this . _logger . verbose ( `Search "${ taskId }" was canceled.` ) ; } ) . catch ( ( searchErr ) => { this . _logger . error ( `Error during search (task: ${ taskId }): ${ searchErr . message }` ) ; reject ( searchErr ) ; } ) . finally ( ( ) => { this . _stream . removeProgressSession ( taskId ) ; this . _stock . search . delete ( taskId ) ; if ( ! isAppending ) { this . _operations . append . setOffset ( this . _operations . search . getOffset ( ) ) ; this . _operations . append . setReadFrom ( this . _operations . search . getReadBytesAmount ( ) ) ; } } ) ; } ) . cancel ( ( ) => { task . break ( ) ; } ) ; } public inspect ( requests ) { if ( this . _stock . inspecting . size !== 0 ) { return new Error ( `Fail to start inspecting because previous wasn't finished.` ) ; } const taskId = guid ( ) ; const measure = this . _logger . measure ( `inspecting` ) ; this . _stream . addProgressSession ( taskId , 'inspecting' ) ; this . _stream . updateProgressSession ( taskId , 0 ) ; const stock = new Map ( ) ; return new CancelablePromise < void , void > ( ( resolve , reject , cancel , self ) => { this . _stock . inspecting . set ( taskId , self ) ; this . _operations . inspecting . setReadTo ( this . _size ) ; requests . forEach ( ( request ) => { const requestTaskId = guid ( ) ; const task = this . _operations . inspecting . perform ( request ) ; stock . set ( requestTaskId , task ) ; task . then ( ( ) => { stock . delete ( requestTaskId ) ; if ( stock . size === 0 ) { return resolve ( undefined ) ; } } ) . catch ( ( error ) => { stock . delete ( requestTaskId ) ; if ( stock . size === 0 ) { return resolve ( undefined ) ; } this . _logger . warn ( `Fail to inspect request "${ request . source }" due error: ${ error . message }` ) ; reject ( error ) ; } ) ; } ) ; self . finally ( ( ) => { stock . forEach ( ( notFinishedTask ) => { notFinishedTask . break ( ) ; } ) ; stock . clear ( ) ; this . _stream . removeProgressSession ( taskId ) ; this . _stock . inspecting . delete ( taskId ) ; measure ( ) ; } ) ; } ) . cancel ( ( ) => { this . _logger . verbose ( `Inspecting was canceled.` ) ; } ) ; } public cancel ( ) { return new Promise ( ( resolve , reject ) => { this . _stock . search . forEach ( ( task ) => { task . break ( ) ; } ) ; this . _stock . inspecting . forEach ( ( task ) => { task . break ( ) ; } ) ; this . _stock . search . clear ( ) ; this . _stock . inspecting . clear ( ) ; Promise . all ( [ this . _operations . append . drop ( ) , this . _operations . inspecting . drop ( ) , this . _operations . search . drop ( ) , ] ) . catch ( ( err ) => { this . _logger . error ( `Error during dropping search operations: ${ err . message }` ) ; } ) . finally ( resolve ) ; } ) ; } public isWorking ( ) { return ( this . _stock . search . size + this . _stock . inspecting . size ) > 0 ; } public getSearchResultMap ( factor , details , range ? ) { return this . _operations . inspecting . getMap ( this . _stream . getStreamLength ( ) , factor , details , range ) ; } public getIndexAround ( position ) { return this . _operations . inspecting . getIndexAround ( position ) ; } private _onMapUpdated ( event ) { this . emit ( SearchEngine . Events . onMapUpdated , event ) ; } }