import * as Path from 'path' ; import * as fs from 'fs' ; import * as Net from 'net' ; import * as FS from '../tools/fs' ; import * as Stream from 'stream' ; import * as Tools from '../tools/index' ; import ServicePaths from './service.paths' ; import ServiceUserPaths from './service.paths.user' ; import ServicePlugins from './service.plugins' ; import ServiceElectron from './service.electron' ; import Logger from '../tools/env.logger' ; import ControllerStreamSearch from '../controllers/stream.search/controller' ; import ControllerStreamCharts from '../controllers/stream.charts/controller' ; import ControllerStreamRanges from '../controllers/stream.ranges/controller' ; import ControllerStreamProcessor from '../controllers/stream.main/controller' ; import ControllerStreamShell from '../controllers/stream.shell/controller' ; import ControllerStreamLogcat from '../controllers/stream.adb/controller' ; import { IPCMessages as IPCElectronMessages , Subscription } from './service.electron' ; import { EventsHub } from '../controllers/stream.common/events' ; import { IService } from '../interfaces/interface.service' ; import { IMapItem } from '../controllers/stream.main/file.map' ; export interface IStreamInfo { guid : string ; socketFile : string ; streamFile : string ; searchFile : string ; bounds : string [ ] ; connections : Net . Socket [ ] ; connectionFactory : ( pluginName : string ) => Promise < { socket : Net . Socket ; file : string ; } > ; server : Net . Server ; events : EventsHub ; processor : ControllerStreamProcessor ; search : ControllerStreamSearch ; charts : ControllerStreamCharts ; ranges : ControllerStreamRanges ; shell : ControllerStreamShell ; logcat : ControllerStreamLogcat ; received : number ; } export interface IPipeOptions { reader : fs . ReadStream ; sourceId : number ; pipeId : string ; streamId ? : string ; decoder ? : Stream . Transform ; } export interface INewSessionEvent { stream : IStreamInfo ; } export interface IServiceSubjects { onSessionChanged : Tools . Subject < string > ; onSessionClosed : Tools . Subject < string > ; onSessionCreated : Tools . Subject < INewSessionEvent > ; } type TGuid = string ; class description class ServiceStreams implements IService { private _logger : Logger = new Logger ( 'ServiceStreams' ) ; private _streams : Map < TGuid , IStreamInfo > = new Map ( ) ; private _activeStreamGuid : string = '' ; private _subscriptions : { [ key ] : Subscription | undefined ; } = { } ; private _subjects : IServiceSubjects = { onSessionChanged : new Tools . Subject ( 'onSessionChanged' ) , onSessionClosed : new Tools . Subject ( 'onSessionClosed' ) , onSessionCreated : new Tools . Subject ( 'onSessionCreated' ) , } ; constructor ( ) { this . _ipc_onStreamSetActive = this . _ipc_onStreamSetActive . bind ( this ) ; this . _ipc_onStreamAdd = this . _ipc_onStreamAdd . bind ( this ) ; this . _ipc_onStreamRemoveRequest = this . _ipc_onStreamRemoveRequest . bind ( this ) ; this . _ipc_onStreamReset = this . _ipc_onStreamReset . bind ( this ) ; } returns public init ( ) { return new Promise ( ( resolve , reject ) => { this . _cleanUp ( ) . then ( resolve ) . catch ( reject ) ; ServiceElectron . IPC . subscribe ( IPCElectronMessages . StreamAddRequest , this . _ipc_onStreamAdd ) . then ( ( subscription ) => { this . _subscriptions . streamAdd = subscription ; } ) . catch ( ( error ) => { this . _logger . warn ( `Fail to subscribe to render event "StreamAddRequest" due error: ${ error . message }. This is not blocked error, loading will be continued.` ) ; } ) ; ServiceElectron . IPC . subscribe ( IPCElectronMessages . StreamRemoveRequest , this . _ipc_onStreamRemoveRequest ) . then ( ( subscription ) => { this . _subscriptions . StreamRemoveRequest = subscription ; } ) . catch ( ( error ) => { this . _logger . warn ( `Fail to subscribe to render event "StreamRemove" due error: ${ error . message }. This is not blocked error, loading will be continued.` ) ; } ) ; ServiceElectron . IPC . subscribe ( IPCElectronMessages . StreamSetActive , this . _ipc_onStreamSetActive ) . then ( ( subscription ) => { this . _subscriptions . StreamSetActive = subscription ; } ) . catch ( ( error ) => { this . _logger . warn ( `Fail to subscribe to render event "StreamSetActive" due error: ${ error . message }. This is not blocked error, loading will be continued.` ) ; } ) ; ServiceElectron . IPC . subscribe ( IPCElectronMessages . StreamResetRequest , this . _ipc_onStreamReset ) . then ( ( subscription ) => { this . _subscriptions . StreamReset = subscription ; } ) . catch ( ( error ) => { this . _logger . warn ( `Fail to subscribe to render event "StreamReset" due error: ${ error . message }. This is not blocked error, loading will be continued.` ) ; } ) ; } ) ; } public getName ( ) { return 'ServiceStreams' ; } public destroy ( ) { return new Promise ( ( resolve ) => { Object . keys ( this . _subscriptions ) . forEach ( ( key ) => { ( this . _subscriptions as any ) [ key ] . destroy ( ) ; } ) ; this . closeAll ( ) . catch ( ( error ) => { this . _logger . warn ( `Fail to close all session due error: ${ error . message }` ) ; } ) . finally ( ( ) => { resolve ( ) ; } ) ; } ) ; } public getSubjects ( ) { return this . _subjects ; } public addStream ( ) { return new Promise ( ( resolve , reject ) => { this . _createStream ( Tools . guid ( ) ) . then ( ( info ) => { resolve ( info . guid ) ; } ) . catch ( ( error ) => { this . _logger . error ( `Fail to create new stream (from backend) due error: ${ error . message }` ) ; reject ( error ) ; } ) ; } ) ; } public writeTo ( chunk , sourceId , trackId , streamId ? ) { return new Promise ( ( resolve , reject ) => { if ( streamId === undefined ) { streamId = this . _activeStreamGuid ; } const stream = this . _streams . get ( streamId ) ; if ( stream === undefined ) { return reject ( new Error ( this . _logger . warn ( `Fail to find a stream data for stream guid "${ streamId }"` ) ) ) ; } stream . processor . write ( chunk , undefined , trackId , sourceId ) . then ( ( ) => { stream . received += chunk . length ; resolve ( ) ; } ) . catch ( ( errorWrite ) => { return reject ( new Error ( this . _logger . warn ( `Fail to process data from stream (${ streamId }) due error: ${ errorWrite . message }` ) ) ) ; } ) ; } ) ; } public getActiveStreamId ( ) { return this . _activeStreamGuid ; } public isStreamExist ( streamId ) { return this . _streams . has ( streamId ) ; } public getStreamFile ( streamId ? ) { if ( streamId === undefined ) { streamId = this . _activeStreamGuid ; } const stream = this . _streams . get ( streamId ) ; if ( stream === undefined ) { return new Error ( this . _logger . warn ( `Cannot return stream's filename, because fail to find a stream data for stream guid "${ streamId }"` ) ) ; } return { streamId : streamId , file : stream . streamFile , bounds : stream . bounds } ; } public getStreamLen ( streamId ? ) { if ( streamId === undefined ) { streamId = this . _activeStreamGuid ; } const stream = this . _streams . get ( streamId ) ; if ( stream === undefined ) { return new Error ( this . _logger . warn ( `Cannot return stream's filename, because fail to find a stream data for stream guid "${ streamId }"` ) ) ; } return stream . processor . getStreamLength ( ) ; } public addProgressSession ( id , name , streamId ? ) { if ( streamId === undefined ) { streamId = this . _activeStreamGuid ; } const stream = this . _streams . get ( streamId ) ; if ( stream === undefined ) { return ; } stream . processor . addProgressSession ( id , name ) ; } public removeProgressSession ( id , streamId ? ) { if ( streamId === undefined ) { streamId = this . _activeStreamGuid ; } const stream = this . _streams . get ( streamId ) ; if ( stream === undefined ) { return ; } stream . processor . removeProgressSession ( id ) ; } public updateProgressSession ( trackId , progress , streamId ? ) { if ( streamId === undefined ) { streamId = this . _activeStreamGuid ; } const stream = this . _streams . get ( streamId ) ; if ( stream === undefined ) { return ; } stream . processor . updateProgressSession ( trackId , progress ) ; } public reattachSessionFileHandle ( streamId ? ) { if ( streamId === undefined ) { streamId = this . _activeStreamGuid ; } const stream = this . _streams . get ( streamId ) ; if ( stream === undefined ) { return ; } stream . processor . reattach ( ) ; } public rewriteStreamFileMap ( streamId , map ) { const stream = this . _streams . get ( streamId ) ; if ( stream === undefined ) { return this . _logger . warn ( `Fail to update stream file map for stream "${ streamId }" because stream doesn't exist.` ) ; } stream . processor . rewriteStreamFileMap ( map ) ; } public pushToStreamFileMap ( streamId , map ) { const stream = this . _streams . get ( streamId ) ; if ( stream === undefined ) { return this . _logger . warn ( `Fail to push stream file map for stream "${ streamId }" because stream doesn't exist.` ) ; } stream . processor . pushToStreamFileMap ( map ) ; } public addBoundFile ( session , filename ) { const stream = this . _streams . get ( session ) ; if ( stream === undefined ) { return ; } stream . bounds . push ( filename ) ; this . _streams . set ( session , stream ) ; } public closeAll ( ) { return new Promise ( ( resolve ) => { Promise . all ( Array . from ( this . _streams . values ( ) ) . map ( ( stream ) => { return this . _destroyStream ( stream . guid ) . catch ( ( error ) => { this . _logger . error ( `Fail destroy stream "${ stream . guid }" due error: ${ error . message }` ) ; return Promise . resolve ( ) ; } ) ; } ) ) . then ( ( ) => { this . _streams . clear ( ) ; this . _cleanUp ( ) . then ( ( ) => { resolve ( ) ; } ) . catch ( ( clearError ) => { this . _logger . warn ( `Fail to cleanup sockets folder due error: ${ clearError . message }` ) ; resolve ( ) ; } ) ; } ) ; } ) ; } returns private _createStream ( guid ) { return new Promise ( ( resolve , reject ) => { const socketFile = this . _getSocketFileName ( `${ Date . now ( ) }-${ guid }` ) ; const streamFile = Path . resolve ( ServiceUserPaths . getStreams ( ) , `${ Date . now ( ) }-${ guid }.stream` ) ; const searchFile = Path . resolve ( ServiceUserPaths . getStreams ( ) , `${ Date . now ( ) }-${ guid }.search` ) ; try { const server = Net . createServer ( this . _acceptConnectionToSocket . bind ( this , guid ) ) ; const events = new EventsHub ( guid ) ; const streamController = new ControllerStreamProcessor ( guid , streamFile , events ) ; const searchController = new ControllerStreamSearch ( guid , streamFile , searchFile , streamController , events ) ; const chartsController = new ControllerStreamCharts ( guid , streamFile , searchFile , streamController , events ) ; const rangesController = new ControllerStreamRanges ( guid , streamFile , streamController ) ; const shellController = new ControllerStreamShell ( guid ) ; const logcatController = new ControllerStreamLogcat ( guid ) ; const stream = { guid : guid , socketFile : socketFile , streamFile : streamFile , searchFile : searchFile , server : server , connections : [ ] , bounds : [ ] , connectionFactory : ( pluginName ) : Promise < { socket : Net . Socket ; file : string ; } > => { return new Promise ( ( resolveConnection ) => { const socket = Net . connect ( socketFile , ( ) => { this . _logger . debug ( `Created new connection UNIX socket: ${ socketFile } for plugin "${ pluginName }".` ) ; resolveConnection ( { socket : socket , file : socketFile } ) ; } ) ; ( socket as any ) . __id = Tools . guid ( ) ; } ) ; } , events : events , processor : streamController , search : searchController , charts : chartsController , ranges : rangesController , shell : shellController , logcat : logcatController , received : 0 , } ; server . listen ( socketFile ) ; server . on ( 'error' , ( error ) => { this . _logger . error ( `Error on socket "${ socketFile }": ${ error . message }` ) ; } ) ; this . _streams . set ( guid , stream ) ; resolve ( stream ) ; } catch ( e ) { reject ( e ) ; } } ) ; } private _acceptConnectionToSocket ( guid , socket ) { const stream = this . _streams . get ( guid ) ; if ( stream === undefined ) { return this . _logger . error ( `Accepted connection to stream "${ guid }", which doesn't exist anymore.` ) ; } const pluginRef = Tools . guid ( ) ; socket . on ( 'data' , this . _stream_onData . bind ( this , guid , pluginRef ) ) ; this . _logger . debug ( `New connection to stream "${ guid }" is accepted. Reference: ${ pluginRef }` ) ; stream . connections . push ( socket ) ; } private _destroyStream ( guid ) { return new Promise ( ( resolve , reject ) => { const stream = this . _streams . get ( guid ) ; if ( stream === undefined ) { this . _logger . warn ( `Was gotten command to destroy stream, but stream wasn't found in storage.` ) ; return ; } const socketFile = stream . socketFile ; const streamFile = stream . streamFile ; const searchFile = stream . searchFile ; stream . connections . forEach ( ( connection ) => { connection . removeAllListeners ( ) ; connection . unref ( ) ; connection . destroy ( ) ; } ) ; stream . server . unref ( ) ; stream . connections = [ ] ; const destroyControllers = ( ) => { return Promise . all ( [ stream . processor . destroy ( ) , stream . search . destroy ( ) , stream . ranges . destroy ( ) , stream . shell . destroy ( ) , stream . logcat . destroy ( ) , ] ) ; } ; const unlinkFile = ( file ) : Promise < void > => { return new Promise ( ( resolveUnlink ) => { fs . exists ( file , ( exists ) => { if ( ! exists ) { this . _logger . debug ( `No need to remove file ${ file } because it wasn't created.` ) ; return resolveUnlink ( ) ; } fs . unlink ( file , ( removeSocketFileError ) => { if ( removeSocketFileError ) { this . _logger . warn ( `Fail to remove stream file ${ file } due error: ${ removeSocketFileError . message }` ) ; } resolveUnlink ( ) ; } ) ; } ) ; } ) ; } ; const unlinkStorageFiles = ( ) => { return Promise . all ( [ unlinkFile ( socketFile ) , unlinkFile ( streamFile ) , unlinkFile ( searchFile ) , ] ) ; } ; destroyControllers ( ) . then ( ( ) => { this . _logger . debug ( `Controllers of stream "${ guid }" are destroyed.` ) ; unlinkStorageFiles ( ) . then ( ( ) => { this . _streams . delete ( guid ) ; resolve ( ) ; } ) . catch ( ( unlinkError ) => { this . _streams . delete ( guid ) ; this . _logger . warn ( `Fail to unlink stream files of stream "${ guid }" due error: ${ unlinkError . message }` ) ; reject ( unlinkError ) ; } ) ; } ) . catch ( ( destroyControllersError ) => { this . _logger . warn ( `Fail to destroy controllers of stream "${ guid }" due error: ${ destroyControllersError . message }.` ) ; reject ( destroyControllersError ) ; } ) ; } ) ; } returns private _cleanUp ( ) { return new Promise ( ( resolve , reject ) => { Promise . all ( [ FS . readFolder ( ServiceUserPaths . getSockets ( ) , FS . EReadingFolderTarget . files ) , FS . readFolder ( ServiceUserPaths . getStreams ( ) , FS . EReadingFolderTarget . files ) , ] ) . then ( ( result ) => { const files = [ ] ; const tm = Date . now ( ) ; files . push ( ... result [ 0 ] . map ( ( file ) => Path . resolve ( ServiceUserPaths . getSockets ( ) , file ) ) ) ; files . push ( ... result [ 1 ] . map ( ( file ) => Path . resolve ( ServiceUserPaths . getStreams ( ) , file ) ) ) ; const queue = files . map ( ( file ) => { return new Promise ( ( resolveUnlink ) => { fs . stat ( file , ( errStat , stats ) => { if ( tm - stats . birthtimeMs < 60 * 1000 * 60 * 24 ) { return resolveUnlink ( ) ; } fs . unlink ( file , ( errorUnlink ) => { if ( errorUnlink ) { this . _logger . warn ( `Fail to remove ${ file } due error: ${ errorUnlink . message }` ) ; } resolveUnlink ( ) ; } ) ; } ) ; } ) ; } ) ; Promise . all ( queue ) . then ( ( ) => { resolve ( ) ; } ) . catch ( reject ) ; } ) . catch ( ( readingError ) => { this . _logger . error ( `Fail to read folder ${ ServicePaths . getPlugins ( ) } or ${ ServiceUserPaths . getStreams ( ) } due error: ${ readingError . message }` ) ; reject ( readingError ) ; } ) ; } ) ; } private _ipc_onStreamAdd ( message , response ) { if ( ! ( message instanceof IPCElectronMessages . StreamAddRequest ) ) { return ; } this . _createStream ( message . guid ) . then ( ( stream ) => { if ( this . _activeStreamGuid === '' ) { this . _activeStreamGuid = stream . guid ; } ServicePlugins . addStream ( stream . guid , stream . connectionFactory ) . then ( ( ) => { this . _subjects . onSessionCreated . emit ( { stream : stream , } ) ; ServiceElectron . updateMenu ( ) ; response ( new IPCElectronMessages . StreamAddResponse ( { guid : message . guid , } ) ) ; } ) . catch ( ( pluginsError ) => { const errMsg = `Fail to create stream due error: ${ pluginsError . message }` ; response ( new IPCElectronMessages . StreamAddResponse ( { guid : message . guid , error : errMsg , } ) ) ; this . _logger . error ( errMsg ) ; } ) ; } ) . catch ( ( streamCreateError ) => { const errMsg = `Fail to create stream due error: ${ streamCreateError . message }` ; response ( new IPCElectronMessages . StreamAddResponse ( { guid : message . guid , error : errMsg , } ) ) ; this . _logger . error ( errMsg ) ; } ) ; } private _ipc_onStreamRemoveRequest ( message , response ) { if ( ! ( message instanceof IPCElectronMessages . StreamRemoveRequest ) ) { return ; } this . _destroyStream ( message . guid ) . then ( ( ) => { ServicePlugins . removedStream ( message . guid ) . then ( ( ) => { this . _subjects . onSessionClosed . emit ( message . guid ) ; ServiceElectron . updateMenu ( ) ; response ( new IPCElectronMessages . StreamRemoveResponse ( { guid : message . guid } ) ) ; } ) . catch ( ( plugingsError ) => { this . _logger . error ( `Fail to correctly destroy session "${ message . guid }" due error: ${ plugingsError . message }.` ) ; response ( new IPCElectronMessages . StreamRemoveResponse ( { guid : message . guid , error : plugingsError . message } ) ) ; } ) ; } ) . catch ( ( destroyError ) => { this . _logger . error ( `Fail to correctly destroy session "${ message . guid }" due error: ${ destroyError . message }.` ) ; response ( new IPCElectronMessages . StreamRemoveResponse ( { guid : message . guid , error : destroyError . message } ) ) ; } ) ; } private _ipc_onStreamSetActive ( message ) { if ( ! ( message instanceof IPCElectronMessages . StreamSetActive ) ) { return ; } if ( this . _activeStreamGuid === message . guid ) { return ; } if ( this . _streams . has ( message . guid ) ) { this . _activeStreamGuid = message . guid ; this . _subjects . onSessionChanged . emit ( this . _activeStreamGuid ) ; this . _logger . debug ( `Active session is set to: ${ this . _activeStreamGuid }` ) ; } else { this . _activeStreamGuid = '' ; this . _subjects . onSessionChanged . emit ( this . _activeStreamGuid ) ; this . _logger . debug ( `Active session is not set to: ${ message . guid } as it isn't a stream realated session` ) ; } ServiceElectron . updateMenu ( ) ; } private _ipc_onStreamReset ( message , response ) { if ( ! ( message instanceof IPCElectronMessages . StreamResetRequest ) ) { return ; } const stream = this . _streams . get ( message . guid ) ; if ( stream === undefined ) { return this . _logger . warn ( `Fail to find a stream data for stream guid "${ message . guid }"` ) ; } Promise . all ( [ stream . processor . reset ( ) , stream . search . reset ( ) , ] ) . then ( ( ) => { this . _logger . debug ( `Session "${ message . guid }" was reset.` ) ; response ( new IPCElectronMessages . StreamResetResponse ( { guid : message . guid , } ) ) ; } ) . catch ( ( error ) => { response ( new IPCElectronMessages . StreamResetResponse ( { guid : message . guid , error : this . _logger . warn ( `Fail to reset session "${ message . guid }" due error: ${ error . message }.` ) , } ) ) ; } ) ; } private _stream_onData ( guid , ref , chunk ) { const stream = this . _streams . get ( guid ) ; if ( stream === undefined ) { return this . _logger . warn ( `Fail to find a stream data for stream guid "${ guid }"` ) ; } stream . processor . write ( chunk , ref , undefined ) . then ( ( ) => { stream . received += chunk . length ; } ) . catch ( ( errorWrite ) => { this . _logger . warn ( `Fail to process data from stream (${ guid }) due error: ${ errorWrite . message }` ) ; } ) ; } private _getSocketFileName ( base ) { if ( process . platform === 'win32' ) { return `\\\\.\\pipe\\${ base }` ; } else { return Path . resolve ( ServiceUserPaths . getSockets ( ) , `${ base }.sock` ) ; } } } export default ( new ServiceStreams ( ) ) ;