import ServiceElectron , { IPCMessages as IPCElectronMessages , Subscription } from '../../services/service.electron' ; import Logger from '../../tools/env.logger' ; import ControllerStreamFileReader from '../stream.main/file.reader' ; import ControllerStreamProcessor from '../stream.main/controller' ; import State from './state' ; import { EventsHub } from '../stream.common/events' ; import { ChartingEngine , TChartData , IMatch , IChartRequest } from './engine/controller' ; import * as Tools from '../../tools/index' ; import { IMapItem } from '../stream.main/file.map' ; export interface IRange { from : number ; to : number ; } export interface IRangeMapItem { rows : IRange ; bytes : IRange ; } export default class ControllerStreamCharts { private _logger : Logger ; private _reader : ControllerStreamFileReader ; private _subscriptions : { [ key ] : Subscription | undefined ; } = { } ; private _state : State ; private _charting : ChartingEngine ; private _events : EventsHub ; private _processor : ControllerStreamProcessor ; private _charts : IChartRequest [ ] = [ ] ; private _pending : { bytesToRead : IRange ; rowOffset : number ; } = { bytesToRead : { from : - 1 , to : - 1 } , rowOffset : - 1 , } ; constructor ( guid , streamFile , searchFile , stream , streamState ) { this . _events = streamState ; this . _processor = stream ; this . _state = new State ( guid , streamFile , searchFile ) ; this . _logger = new Logger ( `ControllerStreamCharts: ${ this . _state . getGuid ( ) }` ) ; this . _charting = new ChartingEngine ( this . _state ) ; this . _reader = new ControllerStreamFileReader ( this . _state . getGuid ( ) , this . _state . getStreamFile ( ) ) ; this . _subscriptions . onStreamBytesMapUpdated = this . _events . getSubject ( ) . onStreamBytesMapUpdated . subscribe ( this . _stream_onUpdate . bind ( this ) ) ; ServiceElectron . IPC . subscribe ( IPCElectronMessages . ChartRequest , this . _ipc_onChartRequest . bind ( this ) ) . then ( ( subscription ) => { this . _subscriptions . ChartRequest = subscription ; } ) . catch ( ( error ) => { this . _logger . warn ( `Fail to subscribe to render event "ChartRequest" due error: ${ error . message }. This is not blocked error, loading will be continued.` ) ; } ) ; ServiceElectron . IPC . subscribe ( IPCElectronMessages . ChartRequestCancelRequest , this . _ipc_onChartRequestCancelRequest . bind ( this ) ) . then ( ( subscription ) => { this . _subscriptions . ChartRequestCancelRequest = subscription ; } ) . catch ( ( error ) => { this . _logger . warn ( `Fail to subscribe to render event "ChartRequestCancelRequest" due error: ${ error . message }. This is not blocked error, loading will be continued.` ) ; } ) ; } public destroy ( ) { return new Promise ( ( resolve , reject ) => { Object . keys ( this . _subscriptions ) . forEach ( ( key ) => { ( this . _subscriptions as any ) [ key ] . destroy ( ) ; } ) ; this . _clear ( ) . catch ( ( error ) => { this . _logger . error ( `Error while killing: ${ error . message }` ) ; } ) . finally ( ( ) => { this . _charting . destroy ( ) ; this . _reader . destroy ( ) ; resolve ( ) ; } ) ; } ) ; } private _extract ( charts , requestId , from ? , to ? , rowOffset ? ) { return new Promise ( ( resolve , reject ) => { if ( this . _processor . getStreamSize ( ) === 0 ) { this . _charts = charts ; return resolve ( { } ) ; } const inspecting = this . _charting . extract ( charts , from , to , rowOffset ) ; if ( inspecting instanceof Error ) { this . _logger . warn ( `Fail to start extract chart data due error: ${ inspecting . message }` ) ; return ; } inspecting . then ( ( data ) => { this . _charts = charts ; resolve ( data ) ; } ) . catch ( ( execErr ) => { reject ( execErr ) ; this . _logger . warn ( `Fail to make extract chart data due error: ${ execErr . message }` ) ; } ) ; } ) ; } private _append ( updated ? ) { if ( this . _charts . length === 0 ) { return ; } if ( updated !== undefined ) { if ( this . _pending . rowOffset === - 1 ) { this . _pending . rowOffset = updated . rows . from ; } if ( this . _pending . bytesToRead . from === - 1 || this . _pending . bytesToRead . from > updated . bytes . from ) { this . _pending . bytesToRead . from = updated . bytes . from ; } if ( this . _pending . bytesToRead . to === - 1 || this . _pending . bytesToRead . to < updated . bytes . to ) { this . _pending . bytesToRead . to = updated . bytes . to ; } } if ( this . _charting . isWorking ( ) ) { return ; } const bytes = { from : this . _pending . bytesToRead . from , to : this . _pending . bytesToRead . to } ; const rowsOffset = this . _pending . rowOffset ; this . _pending . bytesToRead = { from : - 1 , to : - 1 } ; this . _pending . rowOffset = - 1 ; this . _extract ( this . _charts , Tools . guid ( ) , bytes . from , bytes . to , rowsOffset ) . then ( ( data ) => { ServiceElectron . IPC . send ( new IPCElectronMessages . ChartResultsUpdated ( { streamId : this . _state . getGuid ( ) , results : data , } ) ) . catch ( ( sendMsgErr ) => { this . _logger . error ( `Fail notify render due error: ${ sendMsgErr . message }` ) ; } ) ; } ) . catch ( ( searchErr ) => { this . _logger . warn ( `Fail to append search results (range: ${ bytes . from } - ${ bytes . to }) due error: ${ searchErr . message }` ) ; } ) . finally ( ( ) => { this . _reappend ( ) ; } ) ; } private _reappend ( ) { if ( this . _pending . bytesToRead . from !== - 1 && this . _pending . bytesToRead . to !== - 1 ) { this . _append ( ) ; } } private _clear ( ) { return new Promise ( ( resolve , reject ) => { this . _charting . cancel ( ) ; resolve ( ) ; } ) ; } private _ipc_onChartRequest ( message , response ) { const request = message as IPCElectronMessages . ChartRequest ; const started = Date . now ( ) ; if ( this . _state . getGuid ( ) !== request . streamId ) { return ; } if ( request . requests . length === 0 ) { return this . _ipc_chartResultsResponse ( response , { id : request . requestId , started : started , results : { } , } ) ; } this . _clear ( ) . then ( ( ) => { const requests = request . requests . map ( ( regInfo ) => { return { regExp : new RegExp ( regInfo . source , regInfo . flags ) , groups : regInfo . groups , } ; } ) ; this . _extract ( requests , request . requestId ) . then ( ( data ) => { this . _ipc_chartResultsResponse ( response , { id : request . requestId , started : started , results : data , } ) ; } ) . catch ( ( searchErr ) => { return this . _ipc_chartResultsResponse ( response , { id : request . requestId , started : started , error : searchErr . message , } ) ; } ) ; } ) . catch ( ( droppingErr ) => { this . _logger . error ( `Fail drop search file due error: ${ droppingErr . message }` ) ; return this . _ipc_chartResultsResponse ( response , { id : request . requestId , started : started , error : droppingErr . message , } ) ; } ) ; } private _ipc_chartResultsResponse ( response , res ) { response ( new IPCElectronMessages . ChartRequestResults ( { streamId : this . _state . getGuid ( ) , requestId : res . id , error : res . error , results : res . results === undefined ? { } : res . results , duration : Date . now ( ) - res . started , } ) ) ; } private _ipc_onChartRequestCancelRequest ( message , response ) { const request = message as IPCElectronMessages . ChartRequestCancelRequest ; this . _clear ( ) . then ( ( ) => { response ( new IPCElectronMessages . ChartRequestCancelResponse ( { streamId : this . _state . getGuid ( ) , requestId : request . requestId , } ) ) ; } ) . catch ( ( error ) => { response ( new IPCElectronMessages . ChartRequestCancelResponse ( { streamId : this . _state . getGuid ( ) , requestId : request . requestId , error : error . message , } ) ) ; } ) ; } private _stream_onUpdate ( map ) { this . _append ( map ) ; } }