import { PassThrough } from 'stream' ; import miniget from 'miniget' ; import m3u8Parser from './m3u8-parser' ; import DashMPDParser from './dash-mpd-parser' ; import { Callback , Queue } from './queue' ; import { humanStr } from './parse-time' ; import { Item } from './parser' ; namespace m3u8stream { export interface Options { begin ? : number | string ; liveBuffer ? : number ; chunkReadahead ? : number ; highWaterMark ? : number ; requestOptions ? : miniget . Options ; parser ? : 'm3u8' | 'dash-mpd' ; id ? : string ; } export interface Progress { num : number ; size : number ; duration : number ; url : string ; } export interface Stream extends PassThrough { end : ( ) => this ; on ( event , listener ) : this ; on ( event , listener ) : this ; } export interface m3u8streamFunc { ( playlistURL , options ? ) : Stream ; parseTimestamp ( time ) : number ; } } interface TimedItem extends Item { time : number ; } const supportedParsers = { m3u8 : m3u8Parser , 'dash-mpd' : DashMPDParser , } ; let m3u8stream = ( ( playlistURL , options = { } ) : m3u8stream . Stream => { const stream = new PassThrough ( { highWaterMark : options . highWaterMark } ) as m3u8stream . Stream ; const chunkReadahead = options . chunkReadahead || 3 ; const liveBuffer = options . liveBuffer || 20000 ; const requestOptions = options . requestOptions ; const Parser = supportedParsers [ options . parser || ( /\.mpd$/ . test ( playlistURL ) ? 'dash-mpd' : 'm3u8' ) ] ; if ( ! Parser ) { throw TypeError ( `parser '${ options . parser }' not supported` ) ; } let begin = 0 ; if ( typeof options . begin !== 'undefined' ) { begin = typeof options . begin === 'string' ? humanStr ( options . begin ) : Math . max ( options . begin - liveBuffer , 0 ) ; } const forwardEvents = ( req ) => { for ( let event of [ 'abort' , 'request' , 'response' , 'redirect' , 'retry' , 'reconnect' ] ) { req . on ( event , stream . emit . bind ( stream , event ) ) ; } } ; let currSegment ; const streamQueue = new Queue ( ( req , callback ) : void => { currSegment = req ; let size = 0 ; req . on ( 'data' , ( chunk ) => size += chunk . length ) ; req . pipe ( stream , { end : false } ) ; req . on ( 'end' , ( ) => callback ( null , size ) ) ; } , { concurrency : 1 } ) ; let segmentNumber = 0 ; let downloaded = 0 ; const requestQueue = new Queue ( ( segment , callback ) : void => { let reqOptions = Object . assign ( { } , requestOptions ) ; if ( segment . range ) { reqOptions . headers = Object . assign ( { } , reqOptions . headers , { Range : `bytes=${ segment . range . start }-${ segment . range . end }` , } ) ; } let req = miniget ( new URL ( segment . url , playlistURL ) . toString ( ) , reqOptions ) ; req . on ( 'error' , callback ) ; forwardEvents ( req ) ; streamQueue . push ( req , ( _ , size ) => { downloaded += + size ; stream . emit ( 'progress' , { num : ++ segmentNumber , size : size , duration : segment . duration , url : segment . url , } , requestQueue . total , downloaded ) ; callback ( null ) ; } ) ; } , { concurrency : chunkReadahead } ) ; const onError = ( err ) : void => { stream . emit ( 'error' , err ) ; stream . end ( ) ; } ; let refreshThreshold ; let minRefreshTime ; let refreshTimeout ; let fetchingPlaylist = true ; let ended = false ; let isStatic = false ; let lastRefresh ; const onQueuedEnd = ( err ) : void => { currSegment = null ; if ( err ) { onError ( err ) ; } else if ( ! fetchingPlaylist && ! ended && ! isStatic && requestQueue . tasks . length + requestQueue . active <= refreshThreshold ) { let ms = Math . max ( 0 , minRefreshTime - ( Date . now ( ) - lastRefresh ) ) ; fetchingPlaylist = true ; refreshTimeout = setTimeout ( refreshPlaylist , ms ) ; } else if ( ( ended || isStatic ) && ! requestQueue . tasks . length && ! requestQueue . active ) { stream . end ( ) ; } } ; let currPlaylist ; let lastSeq ; let starttime = 0 ; const refreshPlaylist = ( ) : void => { lastRefresh = Date . now ( ) ; currPlaylist = miniget ( playlistURL , requestOptions ) ; currPlaylist . on ( 'error' , onError ) ; forwardEvents ( currPlaylist ) ; const parser = currPlaylist . pipe ( new Parser ( options . id ) ) ; parser . on ( 'starttime' , ( a ) => { if ( starttime ) { return ; } starttime = a ; if ( typeof options . begin === 'string' && begin >= 0 ) { begin += starttime ; } } ) ; parser . on ( 'endlist' , ( ) => { isStatic = true ; } ) ; parser . on ( 'endearly' , currPlaylist . unpipe . bind ( currPlaylist , parser ) ) ; let addedItems = [ ] ; const addItem = ( item ) : void => { if ( ! item . init ) { if ( item . seq <= lastSeq ) { return ; } lastSeq = item . seq ; } begin = item . time ; requestQueue . push ( item , onQueuedEnd ) ; addedItems . push ( item ) ; } ; let tailedItems = [ ] , tailedItemsDuration = 0 ; parser . on ( 'item' , ( item ) => { let timedItem = { time : starttime , ... item } ; if ( begin <= timedItem . time ) { addItem ( timedItem ) ; } else { tailedItems . push ( timedItem ) ; tailedItemsDuration += timedItem . duration ; while ( tailedItems . length > 1 && tailedItemsDuration - tailedItems [ 0 ] . duration > liveBuffer ) { const lastItem = tailedItems . shift ( ) as TimedItem ; tailedItemsDuration -= lastItem . duration ; } } starttime += timedItem . duration ; } ) ; parser . on ( 'end' , ( ) => { currPlaylist = null ; if ( ! addedItems . length && tailedItems . length ) { tailedItems . forEach ( item => { addItem ( item ) ; } ) ; } refreshThreshold = Math . max ( 1 , Math . ceil ( addedItems . length * 0.01 ) ) ; minRefreshTime = addedItems . reduce ( ( total , item ) => item . duration + total , 0 ) ; fetchingPlaylist = false ; onQueuedEnd ( null ) ; } ) ; } ; refreshPlaylist ( ) ; stream . end = ( ) => { ended = true ; streamQueue . die ( ) ; requestQueue . die ( ) ; clearTimeout ( refreshTimeout ) ; currPlaylist ?. destroy ( ) ; currSegment ?. destroy ( ) ; PassThrough . prototype . end . call ( stream , null ) ; return stream ; } ; return stream ; } ) as m3u8stream . m3u8streamFunc ; m3u8stream . parseTimestamp = humanStr ; export = m3u8stream ;