import * as http from 'http' ; import { inject , injectable } from 'inversify' ; import Mirakurun from 'mirakurun' ; import { finished } from 'stream' ; import * as apid from '../../../../api' ; import * as mapid from '../../../../node_modules/mirakurun/api' ; import Reserve from '../../../db/entities/Reserve' ; import Util from '../../../util/Util' ; import IConfigFile from '../../IConfigFile' ; import IConfiguration from '../../IConfiguration' ; import ILogger from '../../ILogger' ; import ILoggerModel from '../../ILoggerModel' ; import IMirakurunClientModel from '../../IMirakurunClientModel' ; import IRecordingStreamCreator from './IRecordingStreamCreator' ; interface TunerProgram { reserve : Reserve ; stream : http . IncomingMessage | null ; } interface TunerStatus { types : mapid . ChannelType [ ] ; programs : TunerProgram [ ] ; } interface TimerIndex { [ key ] : NodeJS . Timeout ; } @ injectable ( ) export default class RecordingStreamCreator implements IRecordingStreamCreator { private log : ILogger ; private config : IConfigFile ; private mirakurunClientModel : IMirakurunClientModel ; private tuners : TunerStatus [ ] = [ ] ; private timerIndex : TimerIndex = { } ; constructor ( @ inject ( 'ILoggerModel' ) logger , @ inject ( 'IConfiguration' ) configuration , @ inject ( 'IMirakurunClientModel' ) mirakurunClientModel ) { this . log = logger . getLogger ( ) ; this . config = configuration . getConfig ( ) ; this . mirakurunClientModel = mirakurunClientModel ; } param tuners public setTuner ( tuners ) { if ( this . tuners . length !== 0 ) { return ; } this . tuners = tuners . map ( tuner => { return { types : tuner . types , programs : [ ] , } ; } ) ; setInterval ( ( ) => { const now = new Date ( ) . getTime ( ) ; for ( const tuner of this . tuners ) { tuner . programs = tuner . programs . filter ( p => { return now - p . reserve . endAt < 12 * 60 * 60 * 1000 ; } ) ; } } , 30 * 60 * 1000 ) ; } param reserveId private deleteReserve ( reserveId ) { clearTimeout ( this . timerIndex [ reserveId ] ) ; delete this . timerIndex [ reserveId ] ; for ( const tuner of this . tuners ) { for ( let i = 0 ; i < tuner . programs . length ; i ++ ) { if ( tuner . programs [ i ] . reserve . id === reserveId ) { const programStream = tuner . programs [ i ] . stream ; if ( programStream !== null ) { programStream . destroy ( ) ; programStream . push ( null ) ; } tuner . programs . splice ( i , 1 ) ; this . log . system . debug ( `delete stream: ${ reserveId }` ) ; return ; } } } } param reserve return public async create ( reserve ) { if ( reserve . isConflict === true ) { return this . getStream ( reserve ) ; } const tunerId = await this . getTunerId ( reserve ) ; if ( tunerId === null ) { this . log . system . warn ( `TunerAssignmentError programId: ${ reserve . id }` ) ; return this . getStream ( reserve ) ; } const stream = this . getStream ( reserve ) ; const tunerProgram = { reserve : reserve , stream : null , } ; this . tuners [ tunerId ] . programs . push ( tunerProgram ) ; try { const s = await stream ; tunerProgram . stream = s ; finished ( tunerProgram . stream , { } , err => { if ( err ) { this . log . system . error ( `RecordingStreamCreator stream error: ${ reserve . id }` ) ; } this . deleteReserve ( reserve . id ) ; } ) ; } catch ( err ) { this . deleteReserve ( reserve . id ) ; } return stream ; } param reserve return private async getTunerId ( reserve ) { for ( let i = 0 ; i < this . tuners . length ; i ++ ) { if ( this . tuners [ i ] . types . indexOf ( < any > reserve . channelType ) !== - 1 && ( this . tuners [ i ] . programs . length === 0 || this . tuners [ i ] . programs [ 0 ] . reserve . channel === reserve . channel ) ) { return i ; } } const now = new Date ( ) . getTime ( ) ; for ( let i = 0 ; i < this . tuners . length ; i ++ ) { if ( this . tuners [ i ] . types . indexOf ( < any > reserve . channelType ) !== - 1 ) { let isOk = true ; for ( const p of this . tuners [ i ] . programs ) { if ( p . reserve . allowEndLack === false || p . reserve . endAt - now > IRecordingStreamCreator . PREP_TIME ) { isOk = false ; break ; } } if ( isOk === false ) { continue ; } const mirakurun = this . mirakurunClientModel . getClient ( ) ; for ( const p of this . tuners [ i ] . programs ) { if ( p . reserve . programId === null ) { continue ; } try { const newProgram = await mirakurun . getProgram ( p . reserve . programId ) ; if ( newProgram . startAt + newProgram . duration - now > IRecordingStreamCreator . PREP_TIME ) { isOk = false ; break ; } } catch ( err ) { this . log . system . warn ( `tuner program get error: ${ p . reserve . id }` ) ; } } if ( isOk === false ) { continue ; } for ( const p of this . tuners [ i ] . programs ) { if ( p . stream !== null ) { p . stream . destroy ( ) ; p . stream . push ( null ) ; } } this . tuners [ i ] . programs = [ ] ; return i ; } } return null ; } param reserve return private getStream ( reserve ) { const mirakurun = this . mirakurunClientModel . getClient ( ) ; mirakurun . priority = reserve . isConflict ? this . config . conflictPriority : this . config . recPriority ; if ( reserve . programId === null ) { return this . getTimeSpecifiedStream ( reserve , mirakurun ) ; } else { return mirakurun . getProgramStream ( reserve . programId , true ) ; } } param reserve param mirakurun return private async getTimeSpecifiedStream ( reserve , mirakurun ) { const now = new Date ( ) . getTime ( ) ; if ( reserve . endAt < now ) { throw new Error ( 'TimeSpecifiedStreamTimeoutError' ) ; } this . timerIndex [ reserve . id ] = setTimeout ( ( ) => { this . destroyStream ( reserve ) ; } , reserve . endAt - now + 1000 * this . config . timeSpecifiedEndMargin ) ; const channelStream = await mirakurun . getServiceStream ( reserve . channelId , true ) . catch ( err => { this . log . system . error ( `stream get error ${ reserve . channelId }` ) ; this . log . system . error ( err ) ; clearTimeout ( this . timerIndex [ reserve . id ] ) ; delete this . timerIndex [ reserve . id ] ; throw err ; } ) ; channelStream . once ( 'end' , ( ) => { clearTimeout ( this . timerIndex [ reserve . id ] ) ; delete this . timerIndex [ reserve . id ] ; } ) ; if ( now < reserve . startAt ) { channelStream . on ( 'data' , ( ) => { } ) ; await Util . sleep ( reserve . startAt - now - 1000 * this . config . timeSpecifiedStartMargin ) ; channelStream . removeAllListeners ( 'data' ) ; } return channelStream ; } param reserve private destroyStream ( reserve ) { clearTimeout ( this . timerIndex [ reserve . id ] ) ; delete this . timerIndex [ reserve . id ] ; let stream = null ; for ( const tuner of this . tuners ) { for ( const program of tuner . programs ) { if ( program . reserve . id === reserve . id ) { stream = program . stream ; } } } if ( stream !== null ) { stream . destroy ( ) ; stream . push ( null ) ; } } param reserve public changeEndAt ( reserve ) { if ( reserve . programId !== null || typeof this . timerIndex [ reserve . id ] === 'undefined' ) { throw new Error ( 'StreamChangeAtError' ) ; } this . timerIndex [ reserve . id ] = setTimeout ( ( ) => { this . destroyStream ( reserve ) ; } , reserve . endAt - new Date ( ) . getTime ( ) + 1000 * this . config . timeSpecifiedEndMargin ) ; } }