import { IncomingMessage } from 'http' ; import { inject , injectable } from 'inversify' ; import mirakurun from 'mirakurun' ; import * as mapid from '../../../node_modules/mirakurun/api' ; import IChannelDB from '../db/IChannelDB' ; import IChannelTypeIndex from '../db/IChannelTypeHash' ; import IProgramDB from '../db/IProgramDB' ; import IConfiguration from '../IConfiguration' ; import ILogger from '../ILogger' ; import ILoggerModel from '../ILoggerModel' ; import IMirakurunClientModel from '../IMirakurunClientModel' ; import IEPGUpdateManageModel , { CreateEvent , ProgramBaseEvent , RedefineEvent , ServiceEvent , } from './IEPGUpdateManageModel' ; @ injectable ( ) class EPGUpdateManageModel implements IEPGUpdateManageModel { private log : ILogger ; private mirakurunClient : mirakurun ; private channelDB : IChannelDB ; private programDB : IProgramDB ; private programQueue : ProgramBaseEvent [ ] = [ ] ; private serviceQueue : ServiceEvent [ ] = [ ] ; private channelIndex : IChannelTypeIndex = { } ; private excludeChannelIndex : { [ channelId ] : boolean ; } = { } ; private excludeSidIndex : { [ serviceId ] : boolean ; } = { } ; constructor ( @ inject ( 'ILoggerModel' ) loggerModel , @ inject ( 'IConfiguration' ) configuration , @ inject ( 'IMirakurunClientModel' ) mirakurunClientModel , @ inject ( 'IChannelDB' ) channelDB , @ inject ( 'IProgramDB' ) programDB ) { this . log = loggerModel . getLogger ( ) ; this . mirakurunClient = mirakurunClientModel . getClient ( ) ; this . channelDB = channelDB ; this . programDB = programDB ; const config = configuration . getConfig ( ) ; if ( typeof config . excludeChannels !== 'undefined' ) { for ( const c of config . excludeChannels ) { this . excludeChannelIndex [ c ] = true ; } } if ( typeof config . excludeSids !== 'undefined' ) { for ( const c of config . excludeSids ) { this . excludeSidIndex [ c ] = true ; } } } /**
     * 番組情報全件更新処理
     */ public async updateAll ( ) { await this . updateChannels ( ) ; const timeout = setTimeout ( ( ) => { this . log . system . error ( 'update all timeout' ) ; clearTimeout ( timeout ) ; throw new Error ( 'EPGUpdateAllTimeoutError' ) ; } , 10 * 60 * 1000 ) ; this . log . system . info ( 'get programs' ) ; const programs = await this . mirakurunClient . getPrograms ( ) . catch ( err => { this . log . system . error ( 'get programs error' ) ; this . log . system . error ( err ) ; clearTimeout ( timeout ) ; throw err ; } ) ; this . log . system . info ( 'done get programs' ) ; const insertPrograms = programs . filter ( p => { return this . isMainProgram ( p ) ; } ) ; this . log . system . info ( 'start update programs' ) ; await this . programDB . insert ( this . channelIndex , insertPrograms ) . catch ( err => { this . log . system . error ( 'update programs error' ) ; this . log . system . error ( err ) ; clearTimeout ( timeout ) ; throw err ; } ) ; this . log . system . info ( 'done update programs' ) ; clearTimeout ( timeout ) ; } param program returns private isMainProgram ( program ) { if ( typeof program . relatedItems === 'undefined' ) { return true ; } for ( const item of program . relatedItems ) { if ( typeof item . type === 'undefined' ) { return true ; } if ( item . type === 'movement' ) { return true ; } if ( item . type === 'relay' ) { continue ; } if ( item . eventId === program . eventId && item . serviceId === program . serviceId ) { return true ; } } return false ; } /**
     * 放送局情報更新
     */ public async updateChannels ( ) { this . log . system . info ( 'get service' ) ; let services = await this . mirakurunClient . getServices ( ) . catch ( err => { this . log . system . error ( 'get service error' ) ; this . log . system . error ( err ) ; throw err ; } ) ; services = services . filter ( s => { return ( typeof this . excludeChannelIndex [ s . id ] === 'undefined' && typeof this . excludeSidIndex [ s . serviceId ] === 'undefined' ) ; } ) ; this . log . system . info ( 'start update channel' ) ; await this . channelDB . insert ( services ) . catch ( err => { this . log . system . error ( 'update channel error' ) ; this . log . system . error ( err ) ; throw err ; } ) ; this . log . system . info ( 'done update channel' ) ; this . channelIndex = { } ; this . updateChannelIndex ( services ) ; } param services return private updateChannelIndex ( services ) { for ( const service of services ) { if ( typeof service . channel === 'undefined' ) { continue ; } if ( typeof this . channelIndex [ service . networkId ] === 'undefined' ) { this . channelIndex [ service . networkId ] = { } ; } this . channelIndex [ service . networkId ] [ service . serviceId ] = { id : service . id , type : service . channel . type , channel : service . channel . channel , } ; } } /**
     * mirakurun の event stream の受信を開始する
     */ public async start ( ) { this . log . system . info ( 'start get stream' ) ; const eventStream = await this . mirakurunClient . getEventsStream ( ) . catch ( err => { this . log . system . error ( 'event stream get error' ) ; this . log . system . error ( err ) ; this . stopStream ( eventStream ) ; throw err ; } ) ; return new Promise < void > ( async ( _resolve , reject ) => { eventStream . once ( 'error' , err => { this . log . system . error ( 'event stream error' ) ; this . log . system . error ( err ) ; this . stopStream ( eventStream ) ; reject ( err ) ; } ) ; eventStream . once ( 'end' , ( ) => { this . log . system . error ( 'event stream is ended' ) ; this . stopStream ( eventStream ) ; reject ( new Error ( 'EndedEventStream' ) ) ; } ) ; eventStream . once ( 'close' , ( ) => { this . log . system . error ( 'event stream is closed' ) ; this . stopStream ( eventStream ) ; reject ( new Error ( 'ClosedEventStream' ) ) ; } ) ; let tmp = Buffer . from ( [ ] ) ; eventStream . on ( 'data' , chunk => { if ( Buffer . compare ( chunk , EPGUpdateManageModel . START_STRING ) === 0 ) { return ; } tmp = Buffer . concat ( [ tmp , chunk ] ) ; if ( Buffer . compare ( tmp . slice ( tmp . length - EPGUpdateManageModel . DATA_DELIMITER_STRING . length , tmp . length ) , EPGUpdateManageModel . DATA_DELIMITER_STRING ) !== 0 ) { return ; } try { this . log . system . debug ( String ( tmp ) ) ; const events = < mapid . Event [ ] > JSON . parse ( `[${ String ( tmp ) . slice ( 0 , - 3 ) }]` ) ; for ( const event of events ) { if ( event . resource === 'program' ) { this . programQueue . push ( < any > event ) ; } else if ( event . resource === 'service' ) { this . serviceQueue . push ( < any > event ) ; } } this . log . system . debug ( 'OK' ) ; } catch ( err ) { this . log . system . error ( 'event stream parse error' ) ; const tmpHex = tmp . toString ( 'hex' ) . match ( /../g ) ; if ( tmpHex !== null ) { this . log . system . debug ( tmpHex . join ( ' ' ) ) ; } this . log . system . error ( err ) ; this . stopStream ( eventStream ) ; reject ( new Error ( 'EventStreamParseError' ) ) ; } tmp = Buffer . from ( [ ] ) ; } ) ; } ) ; } param stream private stopStream ( stream ) { stream . destroy ( ) ; stream . push ( null ) ; stream . removeAllListeners ( ) ; this . programQueue = [ ] ; this . serviceQueue = [ ] ; } return public getProgramQueueSize ( ) { return this . programQueue . length ; } return public getServiceQueueSize ( ) { return this . serviceQueue . length ; } /**
     * programQueue の program を DB へ反映させる
     */ public async saveProgram ( ) { const programs = this . programQueue . splice ( 0 , this . programQueue . length ) ; if ( programs . length === 0 ) { return ; } this . log . system . info ( 'start save program' ) ; const deleteIndex = { } ; const createIndex = { } ; const updateIndex = { } ; for ( const program of programs ) { if ( program . type === 'create' ) { const createData = ( < CreateEvent > program ) . data ; if ( typeof createData . name !== 'undefined' && this . isMainProgram ( createData ) === true ) { createIndex [ createData . id ] = createData ; } } else if ( program . type === 'update' ) { const updateData = ( < CreateEvent > program ) . data ; if ( typeof updateData !== 'undefined' && this . isMainProgram ( updateData ) === true ) { updateIndex [ updateData . id ] = updateData ; } } else if ( program . type === 'remove' || ( program as any ) . type === 'redefine' ) { const from = ( < RedefineEvent > program ) . data . from ; deleteIndex [ from ] = from ; } } const deleteValues = Object . values ( deleteIndex ) ; const insertValues = Object . values ( createIndex ) ; const updateValues = Object . values ( updateIndex ) ; this . log . system . info ( { deleteValues : deleteValues . length , insertValues : insertValues . length , updateValues : updateValues . length , } ) ; this . log . system . info ( 'update db' ) ; await this . programDB . update ( this . channelIndex , { insert : insertValues , update : updateValues , delete : deleteValues , } ) ; this . log . system . info ( 'update db done' ) ; } /**
     * 現在時刻より古い番組情報を削除
     */ public async deleteOldPrograms ( ) { await this . programDB . deleteOld ( new Date ( ) . getTime ( ) ) ; } /**
     * serviceQueue の program を DB へ反映させる
     */ public async saveSevice ( ) { const services = this . serviceQueue . splice ( 0 , this . serviceQueue . length ) ; if ( services . length === 0 ) { return ; } const serviceDatas = await this . mirakurunClient . getServices ( ) . catch ( err => { this . log . system . error ( 'get service error' ) ; this . log . system . error ( err ) ; return [ ] as mapid . Service [ ] ; } ) ; const serviceDataIndex = { } ; for ( const s of serviceDatas ) { serviceDataIndex [ s . id ] = s ; } const createIndex = { } ; const updateIndex = { } ; this . log . system . info ( 'start save service' ) ; for ( const service of services ) { if ( typeof this . excludeChannelIndex [ service . data . id ] !== 'undefined' || typeof this . excludeSidIndex [ service . data . serviceId ] !== 'undefined' ) { continue ; } if ( typeof serviceDataIndex [ service . data . id ] !== 'undefined' ) { service . data . hasLogoData = serviceDataIndex [ service . data . id ] . hasLogoData ; } switch ( service . type ) { case 'create' : if ( typeof service . data . name !== 'undefined' ) { createIndex [ service . data . id ] = service . data ; } break ; case 'update' : if ( typeof service . data !== 'undefined' ) { updateIndex [ service . data . id ] = service . data ; } break ; case 'remove' : break ; } } const insertValues = Object . values ( createIndex ) ; const updateValues = Object . values ( updateIndex ) ; this . log . system . info ( { insertValues : insertValues . length , updateValues : updateValues . length , } ) ; this . log . system . info ( 'update db' ) ; await this . channelDB . update ( { insert : insertValues , update : updateValues , } ) ; this . updateChannelIndex ( insertValues ) ; this . updateChannelIndex ( updateValues ) ; this . log . system . info ( 'update db done' ) ; } } namespace EPGUpdateManageModel { export const START_STRING = Buffer . from ( [ 0x5b , 0x0a ] ) ; export const DATA_DELIMITER_STRING = Buffer . from ( [ 0x7d , 0x0a , 0x2c , 0x0a ] ) ; } export default EPGUpdateManageModel ;