import { IPC } from '../../../../services/service.electron.ipc' ; import { Observable , Subject , Subscription } from 'rxjs' ; import { ControllerSessionTabStreamBookmarks } from '../bookmarks/controller.session.tab.stream.bookmarks' ; import { ControllerSessionScope } from '../scope/controller.session.tab.scope' ; import { ControllerSessionTabTimestamp } from '../timestamps/session.dependency.timestamps' ; import { extractPluginId , extractRowPosition , clearRowStr } from '../../../helpers/row.helpers' ; import { ISelectionAccessor , EParent , } from '../../../../services/standalone/service.output.redirections' ; import { IRow , ControllerRowAPI } from '../row/controller.row.api' ; import { Dependency , SessionGetter } from '../session.dependency' ; import OutputRedirectionsService from '../../../../services/standalone/service.output.redirections' ; import ServiceElectronIpc from '../../../../services/service.electron.ipc' ; import * as Toolkit from 'chipmunk.client.toolkit' ; export type TRequestDataHandler = ( start : number , end : number ) => Promise < IPC . StreamChunk > ; export interface IParamerters { guid : string ; requestDataHandler : TRequestDataHandler ; bookmarks : ControllerSessionTabStreamBookmarks ; scope : ControllerSessionScope ; timestamp : ControllerSessionTabTimestamp ; } export interface IStreamState { count : number ; countRank : number ; stored : IRange ; frame : IRange ; lastLoadingRequestId : any ; bufferLoadingRequestId : any ; } export interface IRange { start : number ; end : number ; } export interface ILoadedRange { range : IRange ; rows : IRow [ ] ; } export interface IPositionData { start : number ; count : number ; } export const Settings = { trigger : 400 , maxRequestCount : 2000 , maxStoredCount : 2000 , requestDelay : 0 , } ; export class ControllerSessionTabStreamOutput implements Dependency { private _guid : string ; private _logger : Toolkit . Logger ; private _rows : IRow [ ] = [ ] ; private _subscriptions : { [ key ] : Toolkit . Subscription | Subscription ; } = { } ; private _preloadTimestamp : number = - 1 ; private _horScrollOffset : number = 0 ; private _lastRequestedRows : IRow [ ] = [ ] ; private _state : IStreamState = { count : 0 , countRank : 1 , stored : { start : - 1 , end : - 1 , } , frame : { start : - 1 , end : - 1 , } , lastLoadingRequestId : undefined , bufferLoadingRequestId : undefined , } ; private _subjects = { onStateUpdated : new Subject < IStreamState > ( ) , onRangeLoaded : new Subject < ILoadedRange > ( ) , onReset : new Subject < void > ( ) , onScrollTo : new Subject < number > ( ) , onSelected : new Subject < number > ( ) , onRankChanged : new Subject < number > ( ) , onSourceChanged : new Subject < number > ( ) , onHorScrollOffset : new Subject < number > ( ) , onPositionChanged : new Subject < IPositionData > ( ) , } ; private _session : SessionGetter ; constructor ( uuid , getter ) { this . _guid = uuid ; this . _session = getter ; this . _logger = new Toolkit . Logger ( `ControllerSessionTabStreamOutput: ${ this . _guid }` ) ; } public init ( ) { return new Promise ( ( resolve ) => { this . _subscriptions . onRowSelected = OutputRedirectionsService . subscribe ( this . _guid , this . _onRowSelected . bind ( this ) ) ; this . _subscriptions . onBookmarkRowSelected = this . _session ( ) . getBookmarks ( ) . getObservable ( ) . onSelected . subscribe ( this . _onRowSelected . bind ( this , EParent . bookmark , undefined ) ) ; resolve ( ) ; } ) ; } public destroy ( ) { return new Promise ( ( resolve , reject ) => { Object . keys ( this . _subscriptions ) . forEach ( ( key ) => { this . _subscriptions [ key ] . unsubscribe ( ) ; } ) ; resolve ( ) ; } ) ; } public getName ( ) { return 'ControllerSessionTabStreamOutput' ; } returns { onStateUpdated public getObservable ( ) { return { onStateUpdated : this . _subjects . onStateUpdated . asObservable ( ) , onRangeLoaded : this . _subjects . onRangeLoaded . asObservable ( ) , onReset : this . _subjects . onReset . asObservable ( ) , onScrollTo : this . _subjects . onScrollTo . asObservable ( ) , onSelected : this . _subjects . onSelected . asObservable ( ) , onRankChanged : this . _subjects . onRankChanged . asObservable ( ) , onSourceChanged : this . _subjects . onSourceChanged . asObservable ( ) , onHorScrollOffset : this . _subjects . onHorScrollOffset . asObservable ( ) , onPositionChanged : this . _subjects . onPositionChanged . asObservable ( ) , } ; } public getFrame ( ) { return { start : this . _state . frame . start >= 0 ? this . _state . frame . start : 0 , end : this . _state . frame . end >= 0 ? this . _state . frame . end : 0 , } ; } public getRange ( range ) { let rows = [ ] ; if ( isNaN ( range . start ) || isNaN ( range . end ) || ! isFinite ( range . start ) || ! isFinite ( range . end ) ) { return new Error ( `Range has incorrect format. Start and end shound be finite and not NaN` ) ; } const stored = Object . assign ( { } , this . _state . stored ) ; if ( this . _state . count === 0 || range . start < 0 || range . end < 0 ) { this . _subjects . onPositionChanged . next ( { start : 0 , count : 0 , } ) ; return [ ] ; } if ( range . start === 0 && range . end === 0 && this . _state . count !== 1 ) { this . _subjects . onPositionChanged . next ( { start : 0 , count : 0 , } ) ; return [ ] ; } if ( stored . start >= 0 && stored . end >= 0 ) { if ( range . start >= stored . start && range . end <= stored . end ) { rows = this . _getRowsSliced ( range . start , range . end + 1 ) ; this . _subjects . onSourceChanged . next ( rows [ 0 ] . pluginId ) ; } else if ( range . end > stored . start && range . start <= stored . start && range . end < stored . end ) { rows = this . _getPendingPackets ( range . start , stored . start ) ; rows . push ( ... this . _getRowsSliced ( stored . start , range . end + 1 ) ) ; } else if ( range . start < stored . end && range . start >= stored . start && range . end > stored . end ) { rows = this . _getPendingPackets ( stored . end + 1 , range . end + 1 ) ; rows . unshift ( ... this . _getRowsSliced ( range . start , stored . end + 1 ) ) ; } else { rows = this . _getPendingPackets ( range . start , range . end + 1 ) ; } } else { rows = this . _getPendingPackets ( range . start , range . end + 1 ) ; } if ( stored . start !== this . _state . stored . start || stored . end !== this . _state . stored . end ) { return new Error ( this . _logger . warn ( `State was changed. Was { start: ${ stored . start }, end: ${ stored . end }}, became { start: ${ this . _state . stored . start }, end: ${ this . _state . stored . end }}.` ) ) ; } if ( rows . length !== range . end - range . start + 1 ) { return new Error ( this . _logger . error ( `Calculation error: gotten ${ rows . length } rows; should be: ${ range . end - range . start + 1 }. State was { start: ${ stored . start }, end: ${ stored . end }}, became { start: ${ this . _state . stored . start }, end: ${ this . _state . stored . end }}.` ) ) ; } this . _state . frame = Object . assign ( { } , range ) ; this . _subjects . onPositionChanged . next ( { start : range . start , count : range . end - range . start , } ) ; return rows ; } public loadRange ( range ) { return new Promise ( ( resolve , reject ) => { if ( isNaN ( range . start ) || isNaN ( range . end ) || ! isFinite ( range . start ) || ! isFinite ( range . end ) ) { return reject ( new Error ( `Range has incorrect format. Start and end shound be finite and not NaN` ) ) ; } const stored = Object . assign ( { } , this . _state . stored ) ; if ( range . start >= stored . start && range . end <= stored . end ) { return resolve ( this . _getRowsSliced ( range . start , range . end + 1 ) ) ; } this . _requestData ( range . start , range . end ) . then ( ( message ) => { const packets = [ ] ; this . _parse ( message . data , message . start , packets ) ; resolve ( packets . filter ( ( packet ) => { return packet . position >= range . start && packet . position <= range . end ; } ) ) ; } ) . catch ( ( error ) => { reject ( new Error ( `Error during requesting data (rows from ${ range . start } to ${ range . end }): ${ error . message }` ) ) ; } ) ; } ) ; } public setFrame ( range ) { if ( this . _state . count === 0 ) { return ; } this . _state . frame = { start : range . start , end : range . end } ; if ( ! this . _load ( ) ) { } } public getRank ( ) { return this . _state . countRank ; } returns public getState ( ) { return Object . assign ( { } , this . _state ) ; } returns public clearStream ( ) { this . _rows = [ ] ; this . _state = { count : 0 , countRank : 1 , stored : { start : - 1 , end : - 1 , } , frame : { start : - 1 , end : - 1 , } , lastLoadingRequestId : undefined , bufferLoadingRequestId : undefined , } ; this . _subjects . onReset . next ( ) ; } param { number } rows returns { IRange | undefined } public updateStreamState ( message ) { this . _setTotalStreamCount ( message . rowsCount ) ; this . _subjects . onStateUpdated . next ( Object . assign ( { } , this . _state ) ) ; } public getBookmarks ( ) { return this . _session ( ) . getBookmarks ( ) ; } public getRowsCount ( ) { return this . _state . count ; } public preload ( range ) { return this . _preload ( range ) ; } public setHorScrollOffset ( offset ) { this . _horScrollOffset = offset ; this . _subjects . onHorScrollOffset . next ( offset ) ; } public getHorScrollOffset ( ) { return this . _horScrollOffset ; } public getRowByPosition ( position ) { return this . _rows . find ( ( row ) => { return row . position === position ; } ) ; } private _isRangeStored ( range ) { const stored = Object . assign ( { } , this . _state . stored ) ; if ( this . _state . count === 0 || range . start < 0 || range . end < 0 ) { return true ; } if ( range . start >= stored . start && range . end <= stored . end ) { return true ; } return false ; } private _getRowsSliced ( from , to ) { const offset = this . _state . stored . start > 0 ? this . _state . stored . start : 0 ; return this . _rows . slice ( from - offset , to - offset ) ; } private _onRowSelected ( sender , selection , clicked ) { if ( sender === EParent . output ) { return ; } this . _subjects . onScrollTo . next ( clicked ) ; } private _load ( ) { clearTimeout ( this . _state . lastLoadingRequestId ) ; const frame = Object . assign ( { } , this . _state . frame ) ; const stored = this . _state . stored ; if ( stored . start <= frame . start && stored . end >= frame . end ) { return false ; } const request = { start : - 1 , end : - 1 } ; if ( frame . end - frame . start + 1 <= 0 ) { return false ; } const distance = { toStart : frame . start , toEnd : this . _state . count - 1 - frame . end , } ; if ( distance . toStart > Settings . maxRequestCount / 2 ) { request . start = distance . toStart - Settings . maxRequestCount / 2 ; } else { request . start = 0 ; } if ( distance . toEnd > Settings . maxRequestCount / 2 ) { request . end = frame . end + Settings . maxRequestCount / 2 ; } else { request . end = this . _state . count - 1 ; } this . _state . lastLoadingRequestId = setTimeout ( ( ) => { this . _state . lastLoadingRequestId = undefined ; this . _requestData ( request . start , request . end ) . then ( ( message ) => { if ( this . _state . lastLoadingRequestId !== undefined ) { this . _lastRequestedRows = [ ] ; this . _parse ( message . data , message . start , this . _lastRequestedRows , frame ) ; return ; } this . _setTotalStreamCount ( message . rows ) ; this . _parse ( message . data , message . start ) ; if ( stored . start <= frame . start && stored . end >= frame . end ) { this . _subjects . onRangeLoaded . next ( { range : { start : frame . start , end : frame . end } , rows : this . _getRowsSliced ( frame . start , frame . end + 1 ) , } ) ; return ; } else { this . _logger . warn ( `Requested frame isn't in scope of stored data. Request - rows from ${ request . start } to ${ request . end }.` ) ; } } ) . catch ( ( error ) => { this . _logger . error ( `Error during requesting data (rows from ${ request . start } to ${ request . end }): ${ error . message }` ) ; } ) ; } , Settings . requestDelay ) ; return true ; } private _preload ( range ) { return new Promise ( ( resolve , reject ) => { const timestamp = Date . now ( ) ; if ( this . _preloadTimestamp !== - 1 && timestamp - this . _preloadTimestamp < 500 ) { return resolve ( null ) ; } if ( this . _isRangeStored ( range ) ) { this . _preloadTimestamp = - 1 ; return resolve ( range ) ; } this . _preloadTimestamp = timestamp ; this . _requestData ( range . start , range . end ) . then ( ( message ) => { this . _preloadTimestamp = - 1 ; this . _setTotalStreamCount ( message . rows ) ; this . _parse ( message . data , range . start ) ; resolve ( { start : message . start , end : message . end } ) ; } ) . catch ( ( error ) => { this . _preloadTimestamp = - 1 ; reject ( new Error ( this . _logger . error ( `Fail to preload data (rows from ${ range . start } to ${ range . end }) due error: ${ error . message }` ) ) ) ; } ) ; } ) ; } param { string } input param { number } start param { number } end param { number } count returns private _parse ( input , start , dest ? , frame ? ) { const rows = input . split ( /\n/gi ) ; let packets = [ ] ; try { rows . forEach ( ( str , i ) => { const position = extractRowPosition ( str ) ; const pluginId = extractPluginId ( str ) ; if ( frame !== undefined ) { if ( frame . end < position ) { throw new Error ( 'No need to parse' ) ; } if ( frame . start > position ) { return ; } } packets . push ( { str : clearRowStr ( str ) , position : position , positionInStream : position , pluginId : pluginId , sessionId : this . _guid , parent : EParent . output , api : this . _session ( ) . getRowAPI ( ) , } ) ; } ) ; } catch ( e ) { } packets = packets . filter ( ( packet ) => { return packet . position !== - 1 ; } ) ; if ( dest !== undefined ) { dest . push ( ... packets ) ; } else { this . _acceptPackets ( packets ) ; } } private _getPendingPackets ( first , last ) { const rows = Array . from ( { length : last - first } ) . map ( ( _ , i ) => { return { pluginId : this . _lastRequestedRows [ i ] === undefined ? - 1 : this . _lastRequestedRows [ i ] . pluginId , position : first + i , positionInStream : first + i , str : this . _lastRequestedRows [ i ] === undefined ? undefined : this . _lastRequestedRows [ i ] . str , rank : this . _state . countRank , sessionId : this . _guid , parent : EParent . output , api : this . _session ( ) . getRowAPI ( ) , } ; } ) ; return rows ; } param { IRow [ ] } packets param { number } start param { number } end returns private _acceptPackets ( packets ) { if ( packets . length === 0 ) { return ; } const size = this . _rows . length ; const packet = { start : packets [ 0 ] . position , end : packets [ packets . length - 1 ] . position , } ; if ( this . _rows . length === 0 ) { this . _rows . push ( ... packets ) ; } else if ( packet . start === this . _state . stored . end + 1 ) { this . _rows . push ( ... packets ) ; if ( this . _rows . length > Settings . maxStoredCount ) { const toCrop = this . _rows . length - Settings . maxStoredCount ; this . _rows . splice ( 0 , toCrop ) ; } } else if ( packet . start > this . _state . stored . end ) { this . _rows = packets ; } else if ( packet . end < this . _state . stored . start ) { this . _rows = packets ; } else if ( packet . start < this . _state . stored . start ) { const range = this . _state . stored . start - packet . start ; const injection = packets . slice ( 0 , range ) ; this . _rows . unshift ( ... injection ) ; if ( this . _rows . length > Settings . maxStoredCount ) { const toCrop = this . _rows . length - Settings . maxStoredCount ; this . _rows . splice ( - toCrop , toCrop ) ; } } else if ( packet . end > this . _state . stored . end ) { const range = packet . end - this . _state . stored . end ; const injection = packets . slice ( packets . length - range , packets . length ) ; this . _rows . push ( ... injection ) ; if ( this . _rows . length > Settings . maxStoredCount ) { const toCrop = this . _rows . length - Settings . maxStoredCount ; this . _rows . splice ( 0 , toCrop ) ; } } this . _state . stored . start = this . _rows [ 0 ] . position ; this . _state . stored . end = this . _rows [ this . _rows . length - 1 ] . position ; if ( size === 0 && this . _rows . length !== 0 ) { this . _subjects . onSourceChanged . next ( this . _rows [ 0 ] . pluginId ) ; } } private _setTotalStreamCount ( count ) { this . _state . count = count ; if ( count === 0 ) { return this . clearStream ( ) ; } const changed = this . _state . countRank !== count . toString ( ) . length ; this . _state . countRank = count . toString ( ) . length ; if ( changed ) { this . _subjects . onRankChanged . next ( this . _state . countRank ) ; } } private _requestData ( start , end ) { return new Promise ( ( resolve , reject ) => { const s = Date . now ( ) ; ServiceElectronIpc . request < IPC . StreamChunk > ( new IPC . StreamChunk ( { guid : this . _guid , start : start , end : end , } ) , IPC . StreamChunk ) . then ( ( response ) => { this . _logger . env ( `Chunk [${ start } - ${ end }] is read in: ${ ( ( Date . now ( ) - s ) / 1000 ) . toFixed ( 2 ) }s` ) ; if ( response . error !== undefined ) { return reject ( new Error ( this . _logger . warn ( `Request to stream chunk was finished within error: ${ response . error }` ) ) ) ; } if ( response . guid === undefined || response . data === undefined || response . length === undefined || response . rows === undefined ) { return reject ( new Error ( this . _logger . warn ( `StreamChunk returns invalid data ${ JSON . stringify ( response ) }` ) ) ) ; } resolve ( { guid : response . guid , data : response . data , start : response . start , end : response . end , length : response . length , rows : response . rows , } ) ; } ) ; } ) ; } }