import registerDebug from 'debug' ; import { Sink } from '../component' ; import { Writable , Readable } from 'stream' ; import { MessageType , Message } from '../message' ; import { isRtcpBye } from '../../utils/protocols/rtcp' ; import { MediaTrack } from '../../utils/protocols/isom' ; const TRIGGER_THRESHOLD = 100 ; const debug = registerDebug ( 'msl:mse' ) ; export class MseSink extends Sink { private readonly _videoEl : HTMLVideoElement ; private _done ? : ( ) => void ; private _lastCheckpointTime : number ; public onSourceOpen ? : ( mse : MediaSource , tracks : MediaTrack [ ] ) => void ; param el constructor ( el ) { if ( el === undefined ) { throw new Error ( 'video element argument missing' ) ; } let mse ; let sourceBuffer ; /**
         * Set up an incoming stream and attach it to the sourceBuffer.
         */ const incoming = new Writable ( { objectMode : true , write : ( msg , _ , callback ) => { if ( msg . type === MessageType . ISOM ) { this . _done = callback ; if ( msg . tracks !== undefined || msg . mime !== undefined ) { const tracks = msg . tracks ?? [ ] ; const mimeCodecs = tracks . map ( ( track ) => track . mime ) . filter ( ( mime ) => mime ) ; const codecs = mimeCodecs . length !== 0 ? mimeCodecs . join ( ', ' ) : 'avc1.640029, mp4a.40.2' ; const mimeType = msg . mime ?? `video/mp4; codecs="${ codecs }"` ; if ( ! MediaSource . isTypeSupported ( mimeType ) ) { incoming . emit ( 'error' , `unsupported media type: ${ mimeType }` ) ; return ; } this . _lastCheckpointTime = 0 ; mse = new MediaSource ( ) ; el . src = window . URL . createObjectURL ( mse ) ; const handler = ( ) => { if ( mse === undefined ) { incoming . emit ( 'error' , 'no MediaSource instance' ) ; return ; } window . URL . revokeObjectURL ( el . src ) ; mse . removeEventListener ( 'sourceopen' , handler ) ; this . onSourceOpen && this . onSourceOpen ( mse , tracks ) ; sourceBuffer = this . addSourceBuffer ( el , mse , mimeType ) ; sourceBuffer . onerror = ( e ) => { console . error ( 'error on SourceBuffer: ' , e ) ; incoming . emit ( 'error' ) ; } ; try { sourceBuffer . appendBuffer ( msg . data ) ; } catch ( err ) { debug ( 'failed to append to SourceBuffer: ' , err , msg ) ; } } ; mse . addEventListener ( 'sourceopen' , handler ) ; } else { this . _lastCheckpointTime = msg . checkpointTime !== undefined ? msg . checkpointTime : this . _lastCheckpointTime ; try { sourceBuffer ?. appendBuffer ( msg . data ) ; } catch ( e ) { debug ( 'failed to append to SourceBuffer: ' , e , msg ) ; } } } else if ( msg . type === MessageType . RTCP ) { if ( isRtcpBye ( msg . rtcp ) ) { mse ?. readyState === 'open' && mse . endOfStream ( ) ; } callback ( ) ; } else { callback ( ) ; } } , } ) ; incoming . on ( 'finish' , ( ) => { console . warn ( 'incoming stream finished: end stream' ) ; mse && mse . readyState === 'open' && mse . endOfStream ( ) ; } ) ; incoming . on ( 'error' , ( msg ) => { console . error ( 'error on incoming stream: ' , msg ) ; if ( sourceBuffer && sourceBuffer . updating ) { sourceBuffer . addEventListener ( 'updateend' , ( ) => { mse ?. readyState === 'open' && mse . endOfStream ( ) ; } ) ; } else { mse ?. readyState === 'open' && mse . endOfStream ( ) ; } } ) ; /**
         * Set up outgoing stream.
         */ const outgoing = new Readable ( { objectMode : true , read : function ( ) { } , } ) ; outgoing . on ( 'error' , ( ) => { console . warn ( 'outgoing stream broke somewhere' ) ; } ) ; /**
         * initialize the component.
         */ super ( incoming , outgoing ) ; this . _videoEl = el ; this . _lastCheckpointTime = 0 ; } param el param mse param mimeType addSourceBuffer ( el , mse , mimeType ) { const sourceBuffer = mse . addSourceBuffer ( mimeType ) ; let trigger = 0 ; const onUpdateEndHandler = ( ) => { ++ trigger ; if ( trigger > TRIGGER_THRESHOLD && sourceBuffer . buffered . length ) { trigger = 0 ; const index = sourceBuffer . buffered . length - 1 ; const start = sourceBuffer . buffered . start ( index ) ; const end = Math . min ( el . currentTime , this . _lastCheckpointTime ) - 10 ; try { if ( end > start ) { sourceBuffer . remove ( start , end ) ; return ; } } catch ( e ) { console . warn ( e ) ; } } this . _done && this . _done ( ) ; } ; sourceBuffer . addEventListener ( 'updateend' , onUpdateEndHandler ) ; return sourceBuffer ; } get currentTime ( ) : number { return this . _videoEl . currentTime ; } async play ( ) { return await this . _videoEl . play ( ) ; } pause ( ) { return this . _videoEl . pause ( ) ; } }