import { inject , injectable } from 'inversify' ; import * as apid from '../../../../../api' ; import IExecutionManagementModel from '../../../IExecutionManagementModel' ; import ILogger from '../../../ILogger' ; import ILoggerModel from '../../../ILoggerModel' ; import ISocketIOManageModel from '../../socketio/ISocketIOManageModel' ; import IStreamBaseModel , { LiveStreamInfo , RecordedStreamInfo } from '../base/IStreamBaseModel' ; import IStreamManageModel , { StreamInfoWithStreamId } from './IStreamManageModel' ; @ injectable ( ) class StreamManageModel implements IStreamManageModel { private log : ILogger ; private executeManagementModel : IExecutionManagementModel ; private socketIO : ISocketIOManageModel ; private streams : { [ streamId ] : IStreamBaseModel < any > ; } = { } ; constructor ( @ inject ( 'ILoggerModel' ) logger , @ inject ( 'IExecutionManagementModel' ) executeManagementModel , @ inject ( 'ISocketIOManageModel' ) socketIO ) { this . log = logger . getLogger ( ) ; this . executeManagementModel = executeManagementModel ; this . socketIO = socketIO ; } param stream return public async start ( stream ) { const exeId = await this . executeManagementModel . getExecution ( StreamManageModel . START_STREAM_PRIORITY ) ; const finalize = ( ) => { this . executeManagementModel . unLockExecution ( exeId ) ; } ; const streamId = this . getEmptyStreamId ( ) ; this . streams [ streamId ] = stream ; this . log . stream . info ( `start stream: ${ streamId . toString ( 10 ) }` ) ; try { await stream . start ( streamId ) ; } catch ( err ) { this . log . stream . error ( 'start stream error' ) ; this . log . stream . error ( err ) ; finalize ( ) ; await this . stop ( streamId ) ; throw err ; } stream . setExitStream ( async ( ) => { finalize ( ) ; await this . stop ( streamId ) . catch ( ) ; } ) ; finalize ( ) ; this . socketIO . notifyClient ( ) ; return streamId ; } return private getEmptyStreamId ( ) { let newStreamId = 0 ; while ( true ) { if ( typeof this . streams [ newStreamId ] === 'undefined' ) { return newStreamId ; } newStreamId ++ ; } } param streamId param isForce return public async stop ( streamId , isForce = false ) { const exeId = await this . executeManagementModel . getExecution ( isForce ? StreamManageModel . FOURCE_STOP_STREAM_PRIORITY : StreamManageModel . STOP_STREAM_PRIORITY ) ; const finalize = ( ) => { this . executeManagementModel . unLockExecution ( exeId ) ; } ; if ( typeof this . streams [ streamId ] === 'undefined' ) { finalize ( ) ; return ; } await this . streams [ streamId ] . stop ( ) . catch ( err => { this . log . stream . error ( `stop stream error ${ streamId }` ) ; finalize ( ) ; throw err ; } ) ; delete this . streams [ streamId ] ; finalize ( ) ; this . socketIO . notifyClient ( ) ; this . log . stream . info ( `stop stream ${ streamId }` ) ; } /**
     * すべてのストリームを停止する
     */ public async stopAll ( ) { for ( const streamId in this . streams ) { await this . stop ( parseInt ( streamId , 10 ) , true ) . catch ( err => { this . log . system . error ( err ) ; } ) ; } } param streamId public keep ( streamId ) { if ( typeof this . streams [ streamId ] === 'undefined' ) { throw new Error ( 'StreamIsUndefined' ) ; } this . streams [ streamId ] . keep ( ) ; } param streamId return public getStreamInfo ( streamId ) { if ( typeof this . streams [ streamId ] === 'undefined' ) { throw new Error ( 'StreamIsNotFound' ) ; } return this . streams [ streamId ] . getInfo ( ) ; } return public getStreamInfos ( ) { const result = [ ] ; for ( const streamId in this . streams ) { result . push ( { streamId : parseInt ( streamId , 10 ) , info : this . streams [ streamId ] . getInfo ( ) , } ) ; } return result ; } } namespace StreamManageModel { export const START_STREAM_PRIORITY = 1 ; export const STOP_STREAM_PRIORITY = 1 ; export const FOURCE_STOP_STREAM_PRIORITY = 10 ; } export default StreamManageModel ;