import * as path from 'path' ; import * as events from 'events' ; import { inject , injectable } from 'inversify' ; import { cloneDeep } from 'lodash' ; import * as apid from '../../../../api' ; import IEncodeEvent from '../../event/IEncodeEvent' ; import IConfiguration from '../../IConfiguration' ; import IExecutionManagementModel from '../../IExecutionManagementModel' ; import ILogger from '../../ILogger' ; import ILoggerModel from '../../ILoggerModel' ; import IEncodeManageModel , { EncodeInfoItem , EncodeQueueInfo , EncodeRecordedIdIndex } from './IEncodeManageModel' ; import { EncodeOption , EncoderModelProvider , IEncoderModel } from './IEncoderModel' ; @ injectable ( ) class EncodeManageModel implements IEncodeManageModel { private log : ILogger ; private executeManagementModel : IExecutionManagementModel ; private encoderModelProvider : EncoderModelProvider ; private encodeEvent : IEncodeEvent ; private concurrentEncodeNum : number ; private waitQueue : IEncoderModel [ ] = [ ] ; private runningQueue : IEncoderModel [ ] = [ ] ; private idCnt : number = 1 ; private listener : events . EventEmitter = new events . EventEmitter ( ) ; constructor ( @ inject ( 'ILoggerModel' ) logger , @ inject ( 'IConfiguration' ) configure , @ inject ( 'IExecutionManagementModel' ) executeManagementModel , @ inject ( 'EncoderModelProvider' ) encoderModelProvider , @ inject ( 'IEncodeEvent' ) encodeEvent ) { this . log = logger . getLogger ( ) ; this . executeManagementModel = executeManagementModel ; this . concurrentEncodeNum = configure . getConfig ( ) . concurrentEncodeNum ; this . encoderModelProvider = encoderModelProvider ; this . encodeEvent = encodeEvent ; this . listener . on ( EncodeManageModel . NEEDS_CHECK_QUEUE_EVENT , this . checkQueue . bind ( this ) ) ; } param addOption return public async push ( addOption ) { if ( this . concurrentEncodeNum <= 0 ) { throw new Error ( 'CncurrentEncodeNumIsZero' ) ; } const exeId = await this . executeManagementModel . getExecution ( EncodeManageModel . ADD_ENCODE_PRIPORITY ) ; const encoder = await this . encoderModelProvider ( ) ; const option = this . createEncodeOption ( addOption ) ; encoder . setOption ( option ) ; this . waitQueue . push ( encoder ) ; this . emitNeedsCheckQueue ( ) ; this . log . encode . info ( `add new encode: ${ option . encodeId }` ) ; this . executeManagementModel . unLockExecution ( exeId ) ; this . encodeEvent . emitAddEncode ( option . encodeId ) ; return option . encodeId ; } param baseOption returns private createEncodeOption ( baseOption ) { const encodeOption = cloneDeep ( baseOption ) as any ; const encodeId = this . idCnt ; encodeOption . encodeId = encodeId ; if ( this . idCnt === Number . MAX_SAFE_INTEGER ) { this . idCnt = 0 ; } this . idCnt ++ ; return encodeOption ; } /**
     * queue の状態をチェックする必要がある場合に呼ぶ
     */ private emitNeedsCheckQueue ( ) { this . listener . emit ( EncodeManageModel . NEEDS_CHECK_QUEUE_EVENT ) ; } return private async checkQueue ( ) { const exeId = await this . executeManagementModel . getExecution ( EncodeManageModel . CREATE_ENCODING_PROCESS_PRIPORITY ) ; if ( this . runningQueue . length >= this . concurrentEncodeNum || this . waitQueue . length === 0 ) { this . executeManagementModel . unLockExecution ( exeId ) ; return ; } const encoder = this . waitQueue . shift ( ) ; if ( typeof encoder === 'undefined' ) { this . executeManagementModel . unLockExecution ( exeId ) ; return ; } const encodeOption = encoder . getEncodeOption ( ) ; if ( encodeOption === null ) { this . executeManagementModel . unLockExecution ( exeId ) ; this . log . encode . warn ( 'encodeOption is null' ) ; return ; } this . runningQueue . push ( encoder ) ; encoder . setOnFinish ( ( isError , outputFilePath ) => { this . onFinish ( isError , outputFilePath , encodeOption ) ; } ) ; let needsFinalize = false ; try { await encoder . start ( ) ; } catch ( err ) { this . log . encode . error ( `create encode process error: ${ encoder . getEncodeId ( ) }` ) ; this . log . encode . error ( err ) ; needsFinalize = true ; this . encodeEvent . emitErrorEncode ( ) ; } this . executeManagementModel . unLockExecution ( exeId ) ; if ( needsFinalize === true ) { this . finalize ( encodeOption . encodeId ) ; } } param isError param outputFilePath param encodeOption private onFinish ( isError , outputFilePath , encodeOption ) { if ( isError ) { this . encodeEvent . emitErrorEncode ( ) ; } else { const fileName = outputFilePath === null ? null : path . basename ( outputFilePath ) ; if ( encodeOption . removeOriginal === true && this . hasSamVideoFileIdItem ( encodeOption . sourceVideoFileId , encodeOption . encodeId ) === true ) { encodeOption . removeOriginal = false ; } this . encodeEvent . emitFinishEncode ( { recordedId : encodeOption . recordedId , videoFileId : encodeOption . sourceVideoFileId , parentDirName : encodeOption . parentDir , filePath : outputFilePath === null || fileName === null ? null : typeof encodeOption . directory === 'undefined' ? fileName : path . join ( encodeOption . directory , fileName ) , fullOutputPath : outputFilePath , mode : encodeOption . mode , removeOriginal : encodeOption . removeOriginal , } ) ; } this . finalize ( encodeOption . encodeId ) ; } param videoFileId param excludeEncodeId return private hasSamVideoFileIdItem ( videoFileId , excludeEncodeId ) { const runningItem = this . runningQueue . find ( i => { const option = i . getEncodeOption ( ) ; return option !== null && option . sourceVideoFileId === videoFileId && option . encodeId !== excludeEncodeId ; } ) ; if ( typeof runningItem !== 'undefined' ) { return true ; } const waitItem = this . waitQueue . find ( i => { const option = i . getEncodeOption ( ) ; return option !== null && option . sourceVideoFileId === videoFileId && option . encodeId !== excludeEncodeId ; } ) ; if ( typeof waitItem !== 'undefined' ) { return true ; } return false ; } param encodeId private async finalize ( encodeId ) { const exeId = await this . executeManagementModel . getExecution ( EncodeManageModel . CLEAR_QUEUE_PRIPORITY ) ; this . runningQueue = this . runningQueue . filter ( q => { return q . getEncodeId ( ) !== encodeId ; } ) ; this . executeManagementModel . unLockExecution ( exeId ) ; process . nextTick ( ( ) => { this . emitNeedsCheckQueue ( ) ; } ) ; } param encodeId public async cancel ( encodeId ) { const exeId = await this . executeManagementModel . getExecution ( EncodeManageModel . CANCEL_ENCODE_PRIPORITY ) ; this . log . encode . info ( `cancel encode: ${ encodeId }` ) ; const runningQueueItem = this . getRunnginQueueItem ( encodeId ) ; if ( typeof runningQueueItem !== 'undefined' ) { await runningQueueItem . cancel ( ) ; } else { this . waitQueue = this . waitQueue . filter ( q => { return q . getEncodeId ( ) !== encodeId ; } ) ; process . nextTick ( ( ) => { this . emitNeedsCheckQueue ( ) ; } ) ; } this . executeManagementModel . unLockExecution ( exeId ) ; this . encodeEvent . emitCancelEncode ( encodeId ) ; } param encodeId return private getRunnginQueueItem ( encodeId ) { return this . runningQueue . find ( q => { return q . getEncodeId ( ) === encodeId ; } ) ; } /**
     * queu に積まれている要素の recorded id の索引を返す
     */ public getRecordedIndex ( ) { const index = { } ; for ( const item of this . runningQueue ) { const itemOption = item . getEncodeOption ( ) ; if ( itemOption === null ) { continue ; } if ( typeof index [ itemOption . recordedId ] === 'undefined' ) { index [ itemOption . recordedId ] = [ ] ; } index [ itemOption . recordedId ] . push ( { encodeId : itemOption . encodeId , name : itemOption . mode , } ) ; } for ( const item of this . waitQueue ) { const itemOption = item . getEncodeOption ( ) ; if ( itemOption === null ) { continue ; } if ( typeof index [ itemOption . recordedId ] === 'undefined' ) { index [ itemOption . recordedId ] = [ ] ; } index [ itemOption . recordedId ] . push ( { encodeId : itemOption . encodeId , name : itemOption . mode , } ) ; } return index ; } param recordedId return public async cancelEncodeByRecordedId ( recordedId ) { const encodeIds = [ ] ; for ( const item of this . waitQueue ) { const itemOption = item . getEncodeOption ( ) ; if ( itemOption === null ) { continue ; } if ( itemOption . recordedId === recordedId ) { encodeIds . push ( itemOption . encodeId ) ; } } for ( const item of this . runningQueue ) { const itemOption = item . getEncodeOption ( ) ; if ( itemOption === null ) { continue ; } if ( itemOption . recordedId === recordedId ) { encodeIds . push ( itemOption . encodeId ) ; } } let isError = false ; for ( const encodeId of encodeIds ) { await this . cancel ( encodeId ) . catch ( err => { isError = true ; this . log . encode . error ( `cancel encode failed: ${ encodeId }` ) ; this . log . encode . error ( err ) ; } ) ; } if ( isError !== false ) { throw new Error ( 'StopEncodeError' ) ; } } return public getEncodeInfo ( ) { const queueInfo = { runningQueue : [ ] , waitQueue : [ ] , } ; for ( const i of this . runningQueue ) { const option = i . getEncodeOption ( ) ; if ( option === null ) { continue ; } const result = { id : option . encodeId , mode : option . mode , recordedId : option . recordedId , } ; const progress = i . getProgressInfo ( ) ; if ( progress !== null ) { result . percent = progress . percent ; result . log = progress . log ; } queueInfo . runningQueue . push ( result ) ; } for ( const i of this . waitQueue ) { const option = i . getEncodeOption ( ) ; if ( option === null ) { continue ; } queueInfo . waitQueue . push ( { id : option . encodeId , mode : option . mode , recordedId : option . recordedId , } ) ; } return queueInfo ; } } namespace EncodeManageModel { export const UNLOCK_EVENT = 'unlockEvent' ; export const UNLOCK_TIMEOUT = 1000 * 60 ; export const CANCEL_ENCODE_PRIPORITY = 1 ; export const ADD_ENCODE_PRIPORITY = 2 ; export const CREATE_ENCODING_PROCESS_PRIPORITY = 2 ; export const CLEAR_QUEUE_PRIPORITY = 3 ; export const NEEDS_CHECK_QUEUE_EVENT = 'needsCheckQueue' ; export const ENCODE_PRIPORITY = 10 ; export const DEFAULT_TIMEOUT_RATE = 4.0 ; } export default EncodeManageModel ;