import { ChildProcess } from 'child_process' ; import * as http from 'http' ; import { inject , injectable } from 'inversify' ; import internal from 'stream' ; import ID3MetadataTransform from 'arib-subtitle-timedmetadater' ; import * as apid from '../../../../../api' ; import ProcessUtil from '../../../../util/ProcessUtil' ; import IConfigFile from '../../../IConfigFile' ; import IConfiguration from '../../../IConfiguration' ; import ILoggerModel from '../../../ILoggerModel' ; import IMirakurunClientModel from '../../../IMirakurunClientModel' ; import IEncodeProcessManageModel , { CreateProcessOption } from '../../encode/IEncodeProcessManageModel' ; import ISocketIOManageModel from '../../socketio/ISocketIOManageModel' ; import IHLSFileDeleterModel from '../util/IHLSFileDeleterModel' ; import ILiveStreamBaseModel , { LiveStreamOption } from './ILiveStreamBaseModel' ; import { LiveStreamInfo } from './IStreamBaseModel' ; import StreamBaseModel from './StreamBaseModel' ; @ injectable ( ) export default abstract class LiveStreamBaseModel extends StreamBaseModel < LiveStreamOption > implements ILiveStreamBaseModel { private stream : http . IncomingMessage | null = null ; private streamProcess : ChildProcess | null = null ; private mirakurunClientModel : IMirakurunClientModel ; private id3MetadataTransoform : ID3MetadataTransform | null = null ; constructor ( @ inject ( 'IConfiguration' ) configure , @ inject ( 'ILoggerModel' ) logger , @ inject ( 'IEncodeProcessManageModel' ) processManager , @ inject ( 'IHLSFileDeleterModel' ) fileDeleter , @ inject ( 'IMirakurunClientModel' ) mirakurunClientModel , @ inject ( 'ISocketIOManageModel' ) socketIO ) { super ( configure , logger , processManager , fileDeleter , socketIO ) ; this . mirakurunClientModel = mirakurunClientModel ; } param streamId return protected createProcessOption ( streamId ) { if ( this . processOption === null ) { throw new Error ( 'ProcessOptionIsNull' ) ; } /**
         * mirakurun の stream をそのまま横流しする
         */ if ( typeof this . processOption . cmd === 'undefined' ) { return null ; } let cmd = this . processOption . cmd . replace ( /%FFMPEG%/g , this . config . ffmpeg ) ; if ( this . getStreamType ( ) === 'LiveHLS' ) { cmd = cmd . replace ( /%streamFileDir%/g , this . config . streamFilePath ) . replace ( /%streamNum%/g , streamId . toString ( 10 ) ) ; } return { input : null , output : this . getStreamType ( ) === 'LiveHLS' ? `${ this . config . streamFilePath }\/stream${ streamId }.m3u8` : null , cmd : cmd , priority : LiveStreamBaseModel . ENCODE_PROCESS_PRIORITY , } ; } param streamId return public async start ( streamId ) { if ( this . processOption === null ) { throw new Error ( 'ProcessOptionIsNull' ) ; } if ( this . getStreamType ( ) === 'LiveHLS' ) { await this . prepStreamDir ( streamId ) ; } await this . setMirakurunStream ( this . config ) ; if ( this . stream === null ) { throw new Error ( 'SetStreamError' ) ; } const poption = this . createProcessOption ( streamId ) ; if ( poption !== null ) { this . log . stream . info ( `create encode process: ${ poption . cmd }` ) ; this . streamProcess = await this . processManager . create ( poption ) . catch ( err => { if ( this . stream !== null ) { this . stream . unpipe ( ) ; this . stream . destroy ( ) ; } this . log . stream . error ( `create encode process failed: ${ poption . cmd }` ) ; throw err ; } ) ; this . streamProcess . on ( 'exit' , ( ) => { this . emitExitStream ( ) ; } ) ; this . streamProcess . on ( 'error' , ( ) => { this . emitExitStream ( ) ; } ) ; if ( this . streamProcess . stderr !== null ) { this . streamProcess . stderr . on ( 'data' , data => { this . log . stream . debug ( String ( data ) ) ; } ) ; } if ( this . streamProcess . stdin !== null ) { if ( this . getStreamType ( ) === 'LiveHLS' ) { this . log . stream . info ( 'use arib-subtitle-timedmetadater' ) ; this . id3MetadataTransoform = new ID3MetadataTransform ( ) ; this . stream . pipe ( this . id3MetadataTransoform ) ; this . id3MetadataTransoform . pipe ( this . streamProcess . stdin ) ; } else { this . stream . pipe ( this . streamProcess . stdin ) ; } } else { await this . stop ( ) ; throw new Error ( 'StreamProcessStdinIsNull' ) ; } if ( this . getStreamType ( ) === 'LiveHLS' ) { this . startCheckStreamEnable ( streamId ) ; } if ( ProcessUtil . isExited ( this . streamProcess ) === true ) { this . streamProcess . removeAllListeners ( ) ; this . emitExitStream ( ) ; } } else { this . stream . on ( 'close' , ( ) => { this . emitExitStream ( ) ; } ) ; this . stream . on ( 'end' , ( ) => { this . emitExitStream ( ) ; } ) ; this . stream . on ( 'error' , ( ) => { this . emitExitStream ( ) ; } ) ; } this . setStopTimer ( ) ; } param config return private async setMirakurunStream ( config ) { if ( this . processOption === null ) { throw new Error ( 'ProcessOptionIsNull' ) ; } const mirakurun = this . mirakurunClientModel . getClient ( ) ; mirakurun . priority = config . streamingPriority ; this . log . stream . info ( `get mirakurun service stream: ${ this . processOption . channelId }` ) ; this . stream = await mirakurun . getServiceStream ( this . processOption . channelId , true , config . streamingPriority ) . catch ( err => { this . stream = null ; if ( this . processOption !== null ) { this . log . system . error ( `get mirakurun service stream failed: ${ this . processOption . channelId }` ) ; } throw err ; } ) ; } return public async stop ( ) { await super . stop ( ) ; if ( this . stream !== null ) { this . stream . unpipe ( ) ; this . stream . destroy ( ) ; } if ( this . id3MetadataTransoform !== null ) { this . id3MetadataTransoform . unpipe ( ) ; this . id3MetadataTransoform . destroy ( ) ; } if ( this . streamProcess !== null ) { await ProcessUtil . kill ( this . streamProcess ) ; } if ( this . getStreamType ( ) === 'LiveHLS' ) { await this . fileDeleter . deleteAllFiles ( ) ; } } return public getStream ( ) { if ( this . streamProcess !== null && this . streamProcess . stdout !== null ) { return this . streamProcess . stdout ; } else if ( this . stream !== null ) { return this . stream ; } else { throw new Error ( 'StreamIsNull' ) ; } } return public getInfo ( ) { if ( this . processOption === null ) { throw new Error ( 'ProcessOptionIsNull' ) ; } if ( this . configMode === null ) { throw new Error ( 'ConfigModeIsNull' ) ; } return { type : this . getStreamType ( ) , mode : this . configMode , channelId : this . processOption . channelId , isEnable : this . isEnable ( ) , } ; } protected abstract getStreamType ( ) ; }