import * as aribts from 'aribts' ; import * as events from 'events' ; import * as fs from 'fs' ; import { inject , injectable } from 'inversify' ; import * as path from 'path' ; import * as stream from 'stream' ; import DateUtil from '../../../util/DateUtil' ; import FileUtil from '../../../util/FileUtil' ; import ILogger from '../../ILogger' ; import ILoggerModel from '../../ILoggerModel' ; import IDropCheckerModel from './IDropCheckerModel' ; @ injectable ( ) class DropCheckerModel implements IDropCheckerModel { private log : ILogger ; private listener : events . EventEmitter = new events . EventEmitter ( ) ; private dest : string | null = null ; private result : aribts . Result | null = null ; private pidIndex : { [ key ] : string ; } = { } ; private time : Date | null = null ; private hasError : boolean = false ; private isFinished : boolean = false ; private transformStream : stream . Transform | null = null ; private tsReadableConnector : aribts . TsReadableConnector | null = null ; private tsPacketParser : aribts . TsPacketParser | null = null ; private tsPacketAnalyzer : aribts . TsPacketAnalyzer | null = null ; private tsSectionParser : aribts . TsSectionParser | null = null ; private tsSectionAnalyzer : aribts . TsSectionAnalyzer | null = null ; private tsSectionUpdater : aribts . TsSectionUpdater | null = null ; private tsPacketSelector : aribts . TsPacketSelector | null = null ; constructor ( @ inject ( 'ILoggerModel' ) logger ) { this . log = logger . getLogger ( ) ; } param logDirPath param srcFilePath param stream return public async start ( logDirPath , srcFilePath , readableStream ) { this . dest = await this . getLogFilePath ( logDirPath , srcFilePath ) ; await FileUtil . touchFile ( this . dest ) ; this . transformStream = new stream . Transform ( { transform : function ( chunk , _encoding , done ) : void { this . push ( chunk ) ; done ( ) ; } , flush : function ( done ) : void { done ( ) ; } , } ) ; this . tsReadableConnector = new aribts . TsReadableConnector ( ) ; this . tsPacketParser = new aribts . TsPacketParser ( ) ; this . tsPacketAnalyzer = new aribts . TsPacketAnalyzer ( ) ; this . tsSectionParser = new aribts . TsSectionParser ( ) ; this . tsSectionAnalyzer = new aribts . TsSectionAnalyzer ( ) ; this . tsSectionUpdater = new aribts . TsSectionUpdater ( ) ; this . tsPacketSelector = new aribts . TsPacketSelector ( { pids : new Array ( 0x30 ) . fill ( 0 ) . map ( ( _ , index ) => index ) , programNumbers : [ ] , } ) ; this . tsSectionUpdater . on ( 'pmt' , tsSection => { const streams = tsSection . decode ( ) . streams ; for ( const s of streams ) { this . setIndex ( s . stream_type , s . elementary_PID ) ; } } ) ; this . tsPacketAnalyzer . on ( 'packetError' , ( pid , counter , expected ) => { this . appendFile ( `error: (pid: ${ this . pidToString ( pid ) }, counter: ${ counter || '-' }, expected: ${ expected || '-' }, time: ${ this . getTime ( ) })\n` ) ; this . hasError = true ; } ) ; this . tsPacketAnalyzer . on ( 'packetDrop' , ( pid , counter , expected ) => { this . appendFile ( `drop (pid: ${ this . pidToString ( pid ) }, counter: ${ counter || '-' }, expected: ${ expected || '-' }, time: ${ this . getTime ( ) })\n` ) ; this . hasError = true ; } ) ; this . tsPacketAnalyzer . on ( 'packetScrambling' , pid => { this . appendFile ( `scrambling (pid: ${ this . pidToString ( pid ) }, time: ${ this . getTime ( ) })\n` ) ; this . hasError = true ; } ) ; this . tsPacketAnalyzer . on ( 'finish' , ( ) => { this . onFinish ( ) ; } ) ; this . tsSectionAnalyzer . on ( 'time' , time => { this . time = time ; } ) ; this . tsSectionParser . on ( 'pmt' , this . tsPacketSelector . onPmt . bind ( this . tsPacketSelector ) ) ; readableStream . pipe ( this . transformStream ) ; this . transformStream . pipe ( this . tsReadableConnector ) ; this . tsReadableConnector . pipe ( this . tsPacketParser as any ) ; this . tsPacketParser . pipe ( this . tsPacketAnalyzer ) ; this . tsPacketParser . pipe ( this . tsSectionParser ) ; this . tsPacketParser . pipe ( this . tsPacketSelector ) ; this . tsSectionParser . pipe ( this . tsSectionAnalyzer ) ; this . tsSectionParser . pipe ( this . tsSectionUpdater ) ; stream . finished ( readableStream , { } , async ( err ) => { if ( err ) { this . log . system . error ( `drop log check stream error: ${ srcFilePath }` ) ; this . stop ( ) ; } } ) ; } returns private async onFinish ( ) { if ( this . isFinished === true ) { return ; } this . isFinished = true ; if ( this . tsPacketAnalyzer === null ) { return ; } this . tsPacketAnalyzer . removeAllListeners ( 'finish' ) ; const result = this . tsPacketAnalyzer . getResult ( ) ; this . result = result ; this . listener . emit ( DropCheckerModel . FINISH_EVENT ) ; if ( this . hasError ) { await this . appendFile ( '\n' ) . catch ( err => { this . log . system . error ( `append error: ${ this . dest }` ) ; this . log . system . error ( err ) ; } ) ; } for ( const pid of Object . keys ( result ) ) { const pidNum = parseInt ( pid , 10 ) ; await this . appendFile ( `pid: ${ this . pidToString ( pidNum ) }, error: ${ result [ pid as any ] . error }, drop: ${ result [ pid as any ] . drop }, scrambling: ${ result [ pid as any ] . scrambling }, packet: ${ result [ pid as any ] . packet }, name: ${ this . getPIDName ( pidNum ) }\n` ) . catch ( err => { this . log . system . error ( `append error: ${ this . dest }` ) ; this . log . system . error ( err ) ; } ) ; } } param logDirPath param srcFilePath param conflict return private async getLogFilePath ( logDirPath , srcFilePath , conflict = 0 ) { let filePath = path . join ( logDirPath , path . basename ( srcFilePath ) ) ; if ( conflict > 0 ) { filePath += `(${ conflict })` ; } filePath += '.log' ; try { await FileUtil . access ( logDirPath , fs . constants . R_OK | fs . constants . W_OK ) ; } catch ( err ) { if ( typeof err . code !== 'undefined' && err . code === 'ENOENT' ) { this . log . system . info ( `mkdirp: ${ logDirPath }` ) ; await FileUtil . mkdir ( logDirPath ) ; } else { this . log . system . fatal ( `dir permission error: ${ logDirPath }` ) ; this . log . system . fatal ( err ) ; throw err ; } } try { await FileUtil . stat ( filePath ) ; return this . getLogFilePath ( logDirPath , srcFilePath , conflict + 1 ) ; } catch ( err ) { return filePath ; } } param streamType param pid private setIndex ( streamType , pid ) { let name ; switch ( streamType ) { case 0x00 : name = 'ECM' ; break ; case 0x02 : name = 'MPEG2 VIDEO' ; break ; case 0x04 : name = 'MPEG2 AUDIO' ; break ; case 0x06 : name = '字幕' ; break ; case 0x0d : name = 'データカルーセル' ; break ; case 0x0f : name = 'MPEG2 AAC' ; break ; case 0x1b : name = 'MPEG4 VIDEO' ; break ; case 0x24 : name = 'HEVC VIDEO' ; break ; default : name = `stream_type 0x${ ( '0000' + pid . toString ( 16 ) ) . slice ( - 4 ) }` ; break ; } this . pidIndex [ pid ] = name ; } param str return private async appendFile ( str ) { if ( this . dest === null ) { throw new Error ( 'LogFilePathIsNull' ) ; } await FileUtil . appendFile ( this . dest , str ) ; } param pid return private pidToString ( pid ) { return '0x' + `${ ( '000' + pid . toString ( 16 ) ) . slice ( - 4 ) }` . toUpperCase ( ) ; } return private getTime ( ) { return this . time === null ? '-' : DateUtil . format ( this . time , 'yyyy/MM/dd hh:mm:ss' ) ; } param pid return private getPIDName ( pid ) { let name ; switch ( pid ) { case 0x0000 : name = 'PAT' ; break ; case 0x0001 : name = 'CAT' ; break ; case 0x0010 : name = 'NIT' ; break ; case 0x0011 : name = 'SDT/BAT' ; break ; case 0x0012 : case 0x0026 : case 0x0027 : name = 'EIT' ; break ; case 0x0013 : name = 'RST' ; break ; case 0x0014 : name = 'TDT/TOT' ; break ; case 0x0017 : name = 'DCT' ; break ; case 0x001e : name = 'DIT' ; break ; case 0x001f : name = 'SIT' ; break ; case 0x0020 : name = 'LIT' ; break ; case 0x0021 : name = 'ERT' ; break ; case 0x0022 : name = 'PCAT' ; break ; case 0x0023 : case 0x0028 : name = 'SDTT' ; break ; case 0x0024 : name = 'BIT' ; break ; case 0x0025 : name = 'NBIT/LDT' ; break ; case 0x0029 : name = 'CDT' ; break ; case 0x1fff : name = 'NULL' ; break ; default : const n = this . pidIndex [ pid ] ; name = typeof n === 'undefined' ? '-' : n ; break ; } return name ; } /**
     * 削除
     */ public async stop ( ) { this . log . system . info ( `stop drop check: ${ this . dest }` ) ; await this . onFinish ( ) . catch ( err => { this . log . system . error ( `finish drop check error: ${ this . dest }` ) ; this . log . system . error ( err ) ; } ) ; if ( this . tsSectionParser !== null ) { this . tsSectionParser . removeAllListeners ( ) ; } if ( this . transformStream !== null ) { this . transformStream . unpipe ( ) ; } if ( this . tsReadableConnector !== null ) { this . tsReadableConnector . removeAllListeners ( ) ; } if ( this . tsPacketParser !== null ) { this . tsPacketParser . removeAllListeners ( ) ; } this . transformStream = null ; this . tsReadableConnector = null ; this . tsPacketParser = null ; this . tsPacketAnalyzer = null ; this . tsSectionParser = null ; this . tsSectionAnalyzer = null ; this . tsSectionUpdater = null ; this . tsPacketSelector = null ; } return public getFilePath ( ) { return this . dest ; } return public async getResult ( ) { if ( this . dest === null ) { throw new Error ( 'DestIsNull' ) ; } await this . setResult ( ) ; if ( this . result === null ) { throw new Error ( 'GetDropResultError' ) ; } return this . result ; } /**
     * finish を待って結果を result へ格納する
     */ private setResult ( ) { return new Promise < void > ( ( resolve , reject ) => { if ( this . result !== null ) { resolve ( ) ; } else { this . listener . once ( DropCheckerModel . FINISH_EVENT , ( ) => { resolve ( ) ; } ) ; setTimeout ( ( ) => { this . listener . removeAllListeners ( ) ; reject ( new Error ( 'GetResultTimeout' ) ) ; } , 10 * 1000 ) ; } } ) ; } } namespace DropCheckerModel { export const FINISH_EVENT = 'finish_event' ; } export default DropCheckerModel ;