2021-08-18 00:01:11 +02:00
const Ffmpeg = require ( 'fluent-ffmpeg' )
const EventEmitter = require ( 'events' )
const Path = require ( 'path' )
const fs = require ( 'fs-extra' )
2021-09-04 21:17:26 +02:00
const Logger = require ( '../Logger' )
const { secondsToTimestamp } = require ( '../utils/fileUtils' )
const { writeConcatFile } = require ( '../utils/ffmpegHelpers' )
const hlsPlaylistGenerator = require ( '../utils/hlsPlaylistGenerator' )
2021-08-18 00:01:11 +02:00
class Stream extends EventEmitter {
constructor ( streamPath , client , audiobook ) {
super ( )
this . id = ( Date . now ( ) + Math . trunc ( Math . random ( ) * 1000 ) ) . toString ( 36 )
this . client = client
this . audiobook = audiobook
this . segmentLength = 6
this . streamPath = Path . join ( streamPath , this . id )
this . concatFilesPath = Path . join ( this . streamPath , 'files.txt' )
this . playlistPath = Path . join ( this . streamPath , 'output.m3u8' )
2021-09-04 21:17:26 +02:00
this . finalPlaylistPath = Path . join ( this . streamPath , 'final-output.m3u8' )
2021-08-18 00:01:11 +02:00
this . startTime = 0
this . ffmpeg = null
this . loop = null
this . isResetting = false
this . isClientInitialized = false
this . isTranscodeComplete = false
this . segmentsCreated = new Set ( )
this . furthestSegmentCreated = 0
this . clientCurrentTime = 0
this . init ( )
}
get socket ( ) {
return this . client . socket
}
get audiobookId ( ) {
return this . audiobook . id
}
2021-09-12 23:10:12 +02:00
get audiobookTitle ( ) {
return this . audiobook ? this . audiobook . title : null
}
2021-08-18 00:01:11 +02:00
get totalDuration ( ) {
return this . audiobook . totalDuration
}
2021-10-01 01:52:32 +02:00
get hlsSegmentType ( ) {
var hasFlac = this . tracks . find ( t => t . ext . toLowerCase ( ) === '.flac' )
return hasFlac ? 'fmp4' : 'mpegts'
}
get segmentBasename ( ) {
if ( this . hlsSegmentType === 'fmp4' ) return 'output-%d.m4s'
return 'output-%d.ts'
}
2021-08-18 00:01:11 +02:00
get segmentStartNumber ( ) {
if ( ! this . startTime ) return 0
return Math . floor ( this . startTime / this . segmentLength )
}
get numSegments ( ) {
var numSegs = Math . floor ( this . totalDuration / this . segmentLength )
if ( this . totalDuration - ( numSegs * this . segmentLength ) > 0 ) {
numSegs ++
}
return numSegs
}
get tracks ( ) {
return this . audiobook . tracks
}
get clientPlaylistUri ( ) {
return ` /hls/ ${ this . id } /output.m3u8 `
}
get clientProgress ( ) {
if ( ! this . clientCurrentTime ) return 0
return Number ( ( this . clientCurrentTime / this . totalDuration ) . toFixed ( 3 ) )
}
toJSON ( ) {
return {
id : this . id ,
clientId : this . client . id ,
userId : this . client . user . id ,
audiobook : this . audiobook . toJSONMinified ( ) ,
segmentLength : this . segmentLength ,
playlistPath : this . playlistPath ,
clientPlaylistUri : this . clientPlaylistUri ,
clientCurrentTime : this . clientCurrentTime ,
startTime : this . startTime ,
2021-08-19 01:31:19 +02:00
segmentStartNumber : this . segmentStartNumber ,
isTranscodeComplete : this . isTranscodeComplete
2021-08-18 00:01:11 +02:00
}
}
init ( ) {
var clientUserAudiobooks = this . client . user ? this . client . user . audiobooks || { } : { }
var userAudiobook = clientUserAudiobooks [ this . audiobookId ] || null
if ( userAudiobook ) {
var timeRemaining = this . totalDuration - userAudiobook . currentTime
2021-10-01 01:52:32 +02:00
Logger . info ( '[STREAM] User has progress for audiobook' , userAudiobook . progress , ` Time Remaining: ${ timeRemaining } s ` )
2021-08-18 00:01:11 +02:00
if ( timeRemaining > 15 ) {
this . startTime = userAudiobook . currentTime
this . clientCurrentTime = this . startTime
}
}
}
async checkSegmentNumberRequest ( segNum ) {
var segStartTime = segNum * this . segmentLength
if ( this . startTime > segStartTime ) {
Logger . warn ( ` [STREAM] Segment # ${ segNum } Request @ ${ secondsToTimestamp ( segStartTime ) } is before start time ( ${ secondsToTimestamp ( this . startTime ) } ) - Reset Transcode ` )
await this . reset ( segStartTime - ( this . segmentLength * 2 ) )
return segStartTime
} else if ( this . isTranscodeComplete ) {
return false
}
var distanceFromFurthestSegment = segNum - this . furthestSegmentCreated
if ( distanceFromFurthestSegment > 10 ) {
Logger . info ( ` Segment # ${ segNum } requested is ${ distanceFromFurthestSegment } segments from latest ( ${ secondsToTimestamp ( segStartTime ) } ) - Reset Transcode ` )
await this . reset ( segStartTime - ( this . segmentLength * 2 ) )
return segStartTime
}
return false
}
updateClientCurrentTime ( currentTime ) {
Logger . debug ( '[Stream] Updated client current time' , secondsToTimestamp ( currentTime ) )
this . clientCurrentTime = currentTime
}
async generatePlaylist ( ) {
fs . ensureDirSync ( this . streamPath )
2021-10-01 01:52:32 +02:00
await hlsPlaylistGenerator ( this . playlistPath , 'output' , this . totalDuration , this . segmentLength , this . hlsSegmentType )
2021-08-18 00:01:11 +02:00
return this . clientPlaylistUri
}
async checkFiles ( ) {
try {
var files = await fs . readdir ( this . streamPath )
files . forEach ( ( file ) => {
var extname = Path . extname ( file )
2021-10-01 01:52:32 +02:00
if ( extname === '.ts' || extname === '.m4s' ) {
2021-08-18 00:01:11 +02:00
var basename = Path . basename ( file , extname )
var num _part = basename . split ( '-' ) [ 1 ]
var part _num = Number ( num _part )
this . segmentsCreated . add ( part _num )
}
} )
if ( ! this . segmentsCreated . size ) {
Logger . warn ( 'No Segments' )
return
}
if ( this . segmentsCreated . size > 6 && ! this . isClientInitialized ) {
this . isClientInitialized = true
Logger . info ( ` [STREAM] ${ this . id } notifying client that stream is ready ` )
this . socket . emit ( 'stream_open' , this . toJSON ( ) )
}
var chunks = [ ]
var current _chunk = [ ]
var last _seg _in _chunk = - 1
var segments = Array . from ( this . segmentsCreated ) . sort ( ( a , b ) => a - b ) ;
var lastSegment = segments [ segments . length - 1 ]
if ( lastSegment > this . furthestSegmentCreated ) {
this . furthestSegmentCreated = lastSegment
}
segments . forEach ( ( seg ) => {
if ( ! current _chunk . length || last _seg _in _chunk + 1 === seg ) {
last _seg _in _chunk = seg
current _chunk . push ( seg )
} else {
if ( current _chunk . length === 1 ) chunks . push ( current _chunk [ 0 ] )
else chunks . push ( ` ${ current _chunk [ 0 ] } - ${ current _chunk [ current _chunk . length - 1 ] } ` )
last _seg _in _chunk = seg
current _chunk = [ seg ]
}
} )
if ( current _chunk . length ) {
if ( current _chunk . length === 1 ) chunks . push ( current _chunk [ 0 ] )
else chunks . push ( ` ${ current _chunk [ 0 ] } - ${ current _chunk [ current _chunk . length - 1 ] } ` )
}
var perc = ( this . segmentsCreated . size * 100 / this . numSegments ) . toFixed ( 2 ) + '%'
Logger . info ( '[STREAM-CHECK] Check Files' , this . segmentsCreated . size , 'of' , this . numSegments , perc , ` Furthest Segment: ${ this . furthestSegmentCreated } ` )
2021-09-22 03:57:33 +02:00
Logger . debug ( '[STREAM-CHECK] Chunks' , chunks . join ( ', ' ) )
2021-08-18 00:01:11 +02:00
this . socket . emit ( 'stream_progress' , {
stream : this . id ,
2021-09-08 16:15:54 +02:00
percent : perc ,
2021-08-18 00:01:11 +02:00
chunks ,
numSegments : this . numSegments
} )
} catch ( error ) {
2021-09-22 03:57:33 +02:00
Logger . error ( 'Failed checking files' , error )
2021-08-18 00:01:11 +02:00
}
}
startLoop ( ) {
2021-09-13 01:22:52 +02:00
// Logger.info(`[Stream] ${this.audiobookTitle} (${this.id}) Start Loop`)
2021-09-08 16:15:54 +02:00
this . socket . emit ( 'stream_progress' , { stream : this . id , chunks : [ ] , numSegments : 0 , percent : '0%' } )
2021-09-13 01:22:52 +02:00
clearInterval ( this . loop )
var intervalId = setInterval ( ( ) => {
2021-08-18 00:01:11 +02:00
if ( ! this . isTranscodeComplete ) {
this . checkFiles ( )
} else {
2021-09-12 23:10:12 +02:00
Logger . info ( ` [Stream] ${ this . audiobookTitle } sending stream_ready ` )
2021-08-18 00:01:11 +02:00
this . socket . emit ( 'stream_ready' )
2021-09-13 01:22:52 +02:00
clearInterval ( intervalId )
2021-08-18 00:01:11 +02:00
}
} , 2000 )
2021-09-13 01:22:52 +02:00
this . loop = intervalId
2021-08-18 00:01:11 +02:00
}
async start ( ) {
Logger . info ( ` [STREAM] START STREAM - Num Segments: ${ this . numSegments } ` )
this . ffmpeg = Ffmpeg ( )
2021-09-04 21:17:26 +02:00
var trackStartTime = await writeConcatFile ( this . tracks , this . concatFilesPath , this . startTime )
2021-08-18 00:01:11 +02:00
this . ffmpeg . addInput ( this . concatFilesPath )
this . ffmpeg . inputFormat ( 'concat' )
this . ffmpeg . inputOption ( '-safe 0' )
if ( this . startTime > 0 ) {
const shiftedStartTime = this . startTime - trackStartTime
Logger . info ( ` [STREAM] Starting Stream at startTime ${ secondsToTimestamp ( this . startTime ) } and Segment # ${ this . segmentStartNumber } ` )
this . ffmpeg . inputOption ( ` -ss ${ shiftedStartTime } ` )
this . ffmpeg . inputOption ( '-noaccurate_seek' )
}
2021-09-08 16:15:54 +02:00
const logLevel = process . env . NODE _ENV === 'production' ? 'error' : 'warning'
2021-10-01 01:52:32 +02:00
const audioCodec = this . hlsSegmentType === 'fmp4' ? 'aac' : 'copy'
2021-08-18 00:01:11 +02:00
this . ffmpeg . addOption ( [
2021-09-08 16:15:54 +02:00
` -loglevel ${ logLevel } ` ,
2021-08-18 00:01:11 +02:00
'-map 0:a' ,
2021-10-01 01:52:32 +02:00
` -c:a ${ audioCodec } `
2021-08-18 00:01:11 +02:00
] )
2021-10-01 01:52:32 +02:00
const hlsOptions = [
2021-08-18 00:01:11 +02:00
'-f hls' ,
"-copyts" ,
"-avoid_negative_ts disabled" ,
"-max_delay 5000000" ,
"-max_muxing_queue_size 2048" ,
` -hls_time 6 ` ,
2021-10-01 01:52:32 +02:00
` -hls_segment_type ${ this . hlsSegmentType } ` ,
2021-08-18 00:01:11 +02:00
` -start_number ${ this . segmentStartNumber } ` ,
"-hls_playlist_type vod" ,
"-hls_list_size 0" ,
"-hls_allow_cache 0"
2021-10-01 01:52:32 +02:00
]
if ( this . hlsSegmentType === 'fmp4' ) {
hlsOptions . push ( '-strict -2' )
var fmp4InitFilename = Path . join ( this . streamPath , 'init.mp4' )
hlsOptions . push ( ` -hls_fmp4_init_filename ${ fmp4InitFilename } ` )
}
this . ffmpeg . addOption ( hlsOptions )
2021-08-18 00:01:11 +02:00
var segmentFilename = Path . join ( this . streamPath , this . segmentBasename )
this . ffmpeg . addOption ( ` -hls_segment_filename ${ segmentFilename } ` )
2021-09-04 21:17:26 +02:00
this . ffmpeg . output ( this . finalPlaylistPath )
2021-08-18 00:01:11 +02:00
this . ffmpeg . on ( 'start' , ( command ) => {
Logger . info ( '[INFO] FFMPEG transcoding started with command: ' + command )
2021-09-13 01:22:52 +02:00
Logger . info ( '' )
2021-08-18 00:01:11 +02:00
if ( this . isResetting ) {
setTimeout ( ( ) => {
Logger . info ( '[STREAM] Clearing isResetting' )
this . isResetting = false
2021-09-13 01:22:52 +02:00
this . startLoop ( )
2021-08-18 00:01:11 +02:00
} , 500 )
2021-09-13 01:22:52 +02:00
} else {
this . startLoop ( )
2021-08-18 00:01:11 +02:00
}
} )
this . ffmpeg . on ( 'stderr' , ( stdErrline ) => {
Logger . info ( stdErrline )
} )
this . ffmpeg . on ( 'error' , ( err , stdout , stderr ) => {
if ( err . message && err . message . includes ( 'SIGKILL' ) ) {
// This is an intentional SIGKILL
Logger . info ( '[FFMPEG] Transcode Killed' )
this . ffmpeg = null
} else {
Logger . error ( 'Ffmpeg Err' , err . message )
}
2021-09-18 01:40:30 +02:00
clearInterval ( this . loop )
2021-08-18 00:01:11 +02:00
} )
this . ffmpeg . on ( 'end' , ( stdout , stderr ) => {
Logger . info ( '[FFMPEG] Transcoding ended' )
2021-08-21 01:29:10 +02:00
// For very small fast load
if ( ! this . isClientInitialized ) {
this . isClientInitialized = true
Logger . info ( ` [STREAM] ${ this . id } notifying client that stream is ready ` )
this . socket . emit ( 'stream_open' , this . toJSON ( ) )
2021-08-18 00:01:11 +02:00
}
this . isTranscodeComplete = true
this . ffmpeg = null
2021-09-18 01:40:30 +02:00
clearInterval ( this . loop )
2021-08-18 00:01:11 +02:00
} )
this . ffmpeg . run ( )
}
async close ( ) {
clearInterval ( this . loop )
Logger . info ( 'Closing Stream' , this . id )
if ( this . ffmpeg ) {
this . ffmpeg . kill ( 'SIGKILL' )
}
await fs . remove ( this . streamPath ) . then ( ( ) => {
Logger . info ( 'Deleted session data' , this . streamPath )
} ) . catch ( ( err ) => {
Logger . error ( 'Failed to delete session data' , err )
} )
this . client . socket . emit ( 'stream_closed' , this . id )
this . emit ( 'closed' )
}
cancelTranscode ( ) {
clearInterval ( this . loop )
if ( this . ffmpeg ) {
this . ffmpeg . kill ( 'SIGKILL' )
}
}
async waitCancelTranscode ( ) {
for ( let i = 0 ; i < 20 ; i ++ ) {
if ( ! this . ffmpeg ) return true
await new Promise ( ( resolve ) => setTimeout ( resolve , 500 ) )
}
Logger . error ( '[STREAM] Transcode never closed...' )
return false
}
async reset ( time ) {
if ( this . isResetting ) {
return Logger . info ( ` [STREAM] Stream ${ this . id } already resetting ` )
}
time = Math . max ( 0 , time )
this . isResetting = true
if ( this . ffmpeg ) {
this . cancelTranscode ( )
await this . waitCancelTranscode ( )
}
this . isTranscodeComplete = false
this . startTime = time
this . clientCurrentTime = this . startTime
Logger . info ( ` Stream Reset New Start Time ${ secondsToTimestamp ( this . startTime ) } ` )
this . start ( )
}
}
module . exports = Stream