server/lib/live/shared/muxing-session.ts

   1 import { mapSeries } from 'bluebird'
   2 import { FSWatcher, watch } from 'chokidar'
   3 import { EventEmitter } from 'events'
   4 import { appendFile, ensureDir, readFile, stat } from 'fs-extra'
   5 import PQueue from 'p-queue'
   6 import { basename, join } from 'path'
   7 import { computeOutputFPS } from '@server/helpers/ffmpeg'
   8 import { logger, loggerTagsFactory, LoggerTagsFn } from '@server/helpers/logger'
   9 import { CONFIG } from '@server/initializers/config'
  10 import { MEMOIZE_TTL, P2P_MEDIA_LOADER_PEER_VERSION, VIDEO_LIVE } from '@server/initializers/constants'
  11 import { removeHLSFileObjectStorageByPath, storeHLSFileFromContent, storeHLSFileFromPath } from '@server/lib/object-storage'
  12 import { VideoFileModel } from '@server/models/video/video-file'
  13 import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist'
  14 import { MStreamingPlaylistVideo, MUserId, MVideoLiveVideo } from '@server/types/models'
  15 import { VideoStorage, VideoStreamingPlaylistType } from '@shared/models'
  16 import {
  17   generateHLSMasterPlaylistFilename,
  18   generateHlsSha256SegmentsFilename,
  19   getLiveDirectory,
  20   getLiveReplayBaseDirectory
  21 } from '../../paths'
  22 import { isAbleToUploadVideo } from '../../user'
  23 import { LiveQuotaStore } from '../live-quota-store'
  24 import { LiveSegmentShaStore } from '../live-segment-sha-store'
  25 import { buildConcatenatedName, getLiveSegmentTime } from '../live-utils'
  26 import { AbstractTranscodingWrapper, FFmpegTranscodingWrapper, RemoteTranscodingWrapper } from './transcoding-wrapper'
  27
  28 import memoizee = require('memoizee')
  29 interface MuxingSessionEvents {
  30   'live-ready': (options: { videoUUID: string }) => void
  31
  32   'bad-socket-health': (options: { videoUUID: string }) => void
  33   'duration-exceeded': (options: { videoUUID: string }) => void
  34   'quota-exceeded': (options: { videoUUID: string }) => void
  35
  36   'transcoding-end': (options: { videoUUID: string }) => void
  37   'transcoding-error': (options: { videoUUID: string }) => void
  38
  39   'after-cleanup': (options: { videoUUID: string }) => void
  40 }
  41
  42 declare interface MuxingSession {
  43   on<U extends keyof MuxingSessionEvents>(
  44     event: U, listener: MuxingSessionEvents[U]
  45   ): this
  46
  47   emit<U extends keyof MuxingSessionEvents>(
  48     event: U, ...args: Parameters<MuxingSessionEvents[U]>
  49   ): boolean
  50 }
  51
  52 class MuxingSession extends EventEmitter {
  53
  54   private transcodingWrapper: AbstractTranscodingWrapper
  55
  56   private readonly context: any
  57   private readonly user: MUserId
  58   private readonly sessionId: string
  59   private readonly videoLive: MVideoLiveVideo
  60
  61   private readonly inputLocalUrl: string
  62   private readonly inputPublicUrl: string
  63
  64   private readonly fps: number
  65   private readonly allResolutions: number[]
  66
  67   private readonly bitrate: number
  68   private readonly ratio: number
  69
  70   private readonly hasAudio: boolean
  71
  72   private readonly videoUUID: string
  73   private readonly saveReplay: boolean
  74
  75   private readonly outDirectory: string
  76   private readonly replayDirectory: string
  77
  78   private readonly lTags: LoggerTagsFn
  79
  80   // Path -> Queue
  81   private readonly objectStorageSendQueues = new Map<string, PQueue>()
  82
  83   private segmentsToProcessPerPlaylist: { [playlistId: string]: string[] } = {}
  84
  85   private streamingPlaylist: MStreamingPlaylistVideo
  86   private liveSegmentShaStore: LiveSegmentShaStore
  87
  88   private filesWatcher: FSWatcher
  89
  90   private masterPlaylistCreated = false
  91   private liveReady = false
  92
  93   private aborted = false
  94
  95   private readonly isAbleToUploadVideoWithCache = memoizee((userId: number) => {
  96     return isAbleToUploadVideo(userId, 1000)
  97   }, { maxAge: MEMOIZE_TTL.LIVE_ABLE_TO_UPLOAD })
  98
  99   private readonly hasClientSocketInBadHealthWithCache = memoizee((sessionId: string) => {
 100     return this.hasClientSocketInBadHealth(sessionId)
 101   }, { maxAge: MEMOIZE_TTL.LIVE_CHECK_SOCKET_HEALTH })
 102
 103   constructor (options: {
 104     context: any
 105     user: MUserId
 106     sessionId: string
 107     videoLive: MVideoLiveVideo
 108
 109     inputLocalUrl: string
 110     inputPublicUrl: string
 111
 112     fps: number
 113     bitrate: number
 114     ratio: number
 115     allResolutions: number[]
 116     hasAudio: boolean
 117   }) {
 118     super()
 119
 120     this.context = options.context
 121     this.user = options.user
 122     this.sessionId = options.sessionId
 123     this.videoLive = options.videoLive
 124
 125     this.inputLocalUrl = options.inputLocalUrl
 126     this.inputPublicUrl = options.inputPublicUrl
 127
 128     this.fps = options.fps
 129
 130     this.bitrate = options.bitrate
 131     this.ratio = options.ratio
 132
 133     this.hasAudio = options.hasAudio
 134
 135     this.allResolutions = options.allResolutions
 136
 137     this.videoUUID = this.videoLive.Video.uuid
 138
 139     this.saveReplay = this.videoLive.saveReplay
 140
 141     this.outDirectory = getLiveDirectory(this.videoLive.Video)
 142     this.replayDirectory = join(getLiveReplayBaseDirectory(this.videoLive.Video), new Date().toISOString())
 143
 144     this.lTags = loggerTagsFactory('live', this.sessionId, this.videoUUID)
 145   }
 146
 147   async runMuxing () {
 148     this.streamingPlaylist = await this.createLivePlaylist()
 149
 150     this.createLiveShaStore()
 151     this.createFiles()
 152
 153     await this.prepareDirectories()
 154
 155     this.transcodingWrapper = this.buildTranscodingWrapper()
 156
 157     this.transcodingWrapper.on('end', () => this.onTranscodedEnded())
 158     this.transcodingWrapper.on('error', () => this.onTranscodingError())
 159
 160     await this.transcodingWrapper.run()
 161
 162     this.filesWatcher = watch(this.outDirectory, { depth: 0 })
 163
 164     this.watchMasterFile()
 165     this.watchTSFiles()
 166   }
 167
 168   abort () {
 169     if (!this.transcodingWrapper) return
 170
 171     this.aborted = true
 172     this.transcodingWrapper.abort()
 173   }
 174
 175   destroy () {
 176     this.removeAllListeners()
 177     this.isAbleToUploadVideoWithCache.clear()
 178     this.hasClientSocketInBadHealthWithCache.clear()
 179   }
 180
 181   private watchMasterFile () {
 182     this.filesWatcher.on('add', async path => {
 183       if (path !== join(this.outDirectory, this.streamingPlaylist.playlistFilename)) return
 184       if (this.masterPlaylistCreated === true) return
 185
 186       try {
 187         if (this.streamingPlaylist.storage === VideoStorage.OBJECT_STORAGE) {
 188           const masterContent = await readFile(path, 'utf-8')
 189           logger.debug('Uploading live master playlist on object storage for %s', this.videoUUID, { masterContent, ...this.lTags() })
 190
 191           const url = await storeHLSFileFromContent(this.streamingPlaylist, this.streamingPlaylist.playlistFilename, masterContent)
 192
 193           this.streamingPlaylist.playlistUrl = url
 194         }
 195
 196         this.streamingPlaylist.assignP2PMediaLoaderInfoHashes(this.videoLive.Video, this.allResolutions)
 197
 198         await this.streamingPlaylist.save()
 199       } catch (err) {
 200         logger.error('Cannot update streaming playlist.', { err, ...this.lTags() })
 201       }
 202
 203       this.masterPlaylistCreated = true
 204
 205       logger.info('Master playlist file for %s has been created', this.videoUUID, this.lTags())
 206     })
 207   }
 208
 209   private watchTSFiles () {
 210     const startStreamDateTime = new Date().getTime()
 211
 212     const addHandler = async (segmentPath: string) => {
 213       if (segmentPath.endsWith('.ts') !== true) return
 214
 215       logger.debug('Live add handler of TS file %s.', segmentPath, this.lTags())
 216
 217       const playlistId = this.getPlaylistIdFromTS(segmentPath)
 218
 219       const segmentsToProcess = this.segmentsToProcessPerPlaylist[playlistId] || []
 220       this.processSegments(segmentsToProcess)
 221
 222       this.segmentsToProcessPerPlaylist[playlistId] = [ segmentPath ]
 223
 224       if (this.hasClientSocketInBadHealthWithCache(this.sessionId)) {
 225         this.emit('bad-socket-health', { videoUUID: this.videoUUID })
 226         return
 227       }
 228
 229       // Duration constraint check
 230       if (this.isDurationConstraintValid(startStreamDateTime) !== true) {
 231         this.emit('duration-exceeded', { videoUUID: this.videoUUID })
 232         return
 233       }
 234
 235       // Check user quota if the user enabled replay saving
 236       if (await this.isQuotaExceeded(segmentPath) === true) {
 237         this.emit('quota-exceeded', { videoUUID: this.videoUUID })
 238       }
 239     }
 240
 241     const deleteHandler = async (segmentPath: string) => {
 242       if (segmentPath.endsWith('.ts') !== true) return
 243
 244       logger.debug('Live delete handler of TS file %s.', segmentPath, this.lTags())
 245
 246       try {
 247         await this.liveSegmentShaStore.removeSegmentSha(segmentPath)
 248       } catch (err) {
 249         logger.warn('Cannot remove segment sha %s from sha store', segmentPath, { err, ...this.lTags() })
 250       }
 251
 252       if (this.streamingPlaylist.storage === VideoStorage.OBJECT_STORAGE) {
 253         try {
 254           await removeHLSFileObjectStorageByPath(this.streamingPlaylist, segmentPath)
 255         } catch (err) {
 256           logger.error('Cannot remove segment %s from object storage', segmentPath, { err, ...this.lTags() })
 257         }
 258       }
 259     }
 260
 261     this.filesWatcher.on('add', p => addHandler(p))
 262     this.filesWatcher.on('unlink', p => deleteHandler(p))
 263   }
 264
 265   private async isQuotaExceeded (segmentPath: string) {
 266     if (this.saveReplay !== true) return false
 267     if (this.aborted) return false
 268
 269     try {
 270       const segmentStat = await stat(segmentPath)
 271
 272       LiveQuotaStore.Instance.addQuotaTo(this.user.id, this.sessionId, segmentStat.size)
 273
 274       const canUpload = await this.isAbleToUploadVideoWithCache(this.user.id)
 275
 276       return canUpload !== true
 277     } catch (err) {
 278       logger.error('Cannot stat %s or check quota of %d.', segmentPath, this.user.id, { err, ...this.lTags() })
 279     }
 280   }
 281
 282   private createFiles () {
 283     for (let i = 0; i < this.allResolutions.length; i++) {
 284       const resolution = this.allResolutions[i]
 285
 286       const file = new VideoFileModel({
 287         resolution,
 288         size: -1,
 289         extname: '.ts',
 290         infoHash: null,
 291         fps: this.fps,
 292         storage: this.streamingPlaylist.storage,
 293         videoStreamingPlaylistId: this.streamingPlaylist.id
 294       })
 295
 296       VideoFileModel.customUpsert(file, 'streaming-playlist', null)
 297         .catch(err => logger.error('Cannot create file for live streaming.', { err, ...this.lTags() }))
 298     }
 299   }
 300
 301   private async prepareDirectories () {
 302     await ensureDir(this.outDirectory)
 303
 304     if (this.videoLive.saveReplay === true) {
 305       await ensureDir(this.replayDirectory)
 306     }
 307   }
 308
 309   private isDurationConstraintValid (streamingStartTime: number) {
 310     const maxDuration = CONFIG.LIVE.MAX_DURATION
 311     // No limit
 312     if (maxDuration < 0) return true
 313
 314     const now = new Date().getTime()
 315     const max = streamingStartTime + maxDuration
 316
 317     return now <= max
 318   }
 319
 320   private processSegments (segmentPaths: string[]) {
 321     mapSeries(segmentPaths, previousSegment => this.processSegment(previousSegment))
 322       .catch(err => {
 323         if (this.aborted) return
 324
 325         logger.error('Cannot process segments', { err, ...this.lTags() })
 326       })
 327   }
 328
 329   private async processSegment (segmentPath: string) {
 330     // Add sha hash of previous segments, because ffmpeg should have finished generating them
 331     await this.liveSegmentShaStore.addSegmentSha(segmentPath)
 332
 333     if (this.saveReplay) {
 334       await this.addSegmentToReplay(segmentPath)
 335     }
 336
 337     if (this.streamingPlaylist.storage === VideoStorage.OBJECT_STORAGE) {
 338       try {
 339         await storeHLSFileFromPath(this.streamingPlaylist, segmentPath)
 340
 341         await this.processM3U8ToObjectStorage(segmentPath)
 342       } catch (err) {
 343         logger.error('Cannot store TS segment %s in object storage', segmentPath, { err, ...this.lTags() })
 344       }
 345     }
 346
 347     // Master playlist and segment JSON file are created, live is ready
 348     if (this.masterPlaylistCreated && !this.liveReady) {
 349       this.liveReady = true
 350
 351       this.emit('live-ready', { videoUUID: this.videoUUID })
 352     }
 353   }
 354
 355   private async processM3U8ToObjectStorage (segmentPath: string) {
 356     const m3u8Path = join(this.outDirectory, this.getPlaylistNameFromTS(segmentPath))
 357
 358     logger.debug('Process M3U8 file %s.', m3u8Path, this.lTags())
 359
 360     const segmentName = basename(segmentPath)
 361
 362     const playlistContent = await readFile(m3u8Path, 'utf-8')
 363     // Remove new chunk references, that will be processed later
 364     const filteredPlaylistContent = playlistContent.substring(0, playlistContent.lastIndexOf(segmentName) + segmentName.length) + '\n'
 365
 366     try {
 367       if (!this.objectStorageSendQueues.has(m3u8Path)) {
 368         this.objectStorageSendQueues.set(m3u8Path, new PQueue({ concurrency: 1 }))
 369       }
 370
 371       const queue = this.objectStorageSendQueues.get(m3u8Path)
 372       await queue.add(() => storeHLSFileFromContent(this.streamingPlaylist, m3u8Path, filteredPlaylistContent))
 373     } catch (err) {
 374       logger.error('Cannot store in object storage m3u8 file %s', m3u8Path, { err, ...this.lTags() })
 375     }
 376   }
 377
 378   private onTranscodingError () {
 379     this.emit('transcoding-error', ({ videoUUID: this.videoUUID }))
 380   }
 381
 382   private onTranscodedEnded () {
 383     this.emit('transcoding-end', ({ videoUUID: this.videoUUID }))
 384
 385     logger.info('RTMP transmuxing for video %s ended. Scheduling cleanup', this.inputLocalUrl, this.lTags())
 386
 387     setTimeout(() => {
 388       // Wait latest segments generation, and close watchers
 389
 390       const promise = this.filesWatcher?.close() || Promise.resolve()
 391       promise
 392         .then(() => {
 393           // Process remaining segments hash
 394           for (const key of Object.keys(this.segmentsToProcessPerPlaylist)) {
 395             this.processSegments(this.segmentsToProcessPerPlaylist[key])
 396           }
 397         })
 398         .catch(err => {
 399           logger.error(
 400             'Cannot close watchers of %s or process remaining hash segments.', this.outDirectory,
 401             { err, ...this.lTags() }
 402           )
 403         })
 404
 405       this.emit('after-cleanup', { videoUUID: this.videoUUID })
 406     }, 1000)
 407   }
 408
 409   private hasClientSocketInBadHealth (sessionId: string) {
 410     const rtmpSession = this.context.sessions.get(sessionId)
 411
 412     if (!rtmpSession) {
 413       logger.warn('Cannot get session %s to check players socket health.', sessionId, this.lTags())
 414       return
 415     }
 416
 417     for (const playerSessionId of rtmpSession.players) {
 418       const playerSession = this.context.sessions.get(playerSessionId)
 419
 420       if (!playerSession) {
 421         logger.error('Cannot get player session %s to check socket health.', playerSession, this.lTags())
 422         continue
 423       }
 424
 425       if (playerSession.socket.writableLength > VIDEO_LIVE.MAX_SOCKET_WAITING_DATA) {
 426         return true
 427       }
 428     }
 429
 430     return false
 431   }
 432
 433   private async addSegmentToReplay (segmentPath: string) {
 434     const segmentName = basename(segmentPath)
 435     const dest = join(this.replayDirectory, buildConcatenatedName(segmentName))
 436
 437     try {
 438       const data = await readFile(segmentPath)
 439
 440       await appendFile(dest, data)
 441     } catch (err) {
 442       logger.error('Cannot copy segment %s to replay directory.', segmentPath, { err, ...this.lTags() })
 443     }
 444   }
 445
 446   private async createLivePlaylist (): Promise<MStreamingPlaylistVideo> {
 447     const playlist = await VideoStreamingPlaylistModel.loadOrGenerate(this.videoLive.Video)
 448
 449     playlist.playlistFilename = generateHLSMasterPlaylistFilename(true)
 450     playlist.segmentsSha256Filename = generateHlsSha256SegmentsFilename(true)
 451
 452     playlist.p2pMediaLoaderPeerVersion = P2P_MEDIA_LOADER_PEER_VERSION
 453     playlist.type = VideoStreamingPlaylistType.HLS
 454
 455     playlist.storage = CONFIG.OBJECT_STORAGE.ENABLED
 456       ? VideoStorage.OBJECT_STORAGE
 457       : VideoStorage.FILE_SYSTEM
 458
 459     return playlist.save()
 460   }
 461
 462   private createLiveShaStore () {
 463     this.liveSegmentShaStore = new LiveSegmentShaStore({
 464       videoUUID: this.videoLive.Video.uuid,
 465       sha256Path: join(this.outDirectory, this.streamingPlaylist.segmentsSha256Filename),
 466       streamingPlaylist: this.streamingPlaylist,
 467       sendToObjectStorage: CONFIG.OBJECT_STORAGE.ENABLED
 468     })
 469   }
 470
 471   private buildTranscodingWrapper () {
 472     const options = {
 473       streamingPlaylist: this.streamingPlaylist,
 474       videoLive: this.videoLive,
 475
 476       lTags: this.lTags,
 477
 478       sessionId: this.sessionId,
 479       inputLocalUrl: this.inputLocalUrl,
 480       inputPublicUrl: this.inputPublicUrl,
 481
 482       toTranscode: this.allResolutions.map(resolution => ({
 483         resolution,
 484         fps: computeOutputFPS({ inputFPS: this.fps, resolution })
 485       })),
 486
 487       fps: this.fps,
 488       bitrate: this.bitrate,
 489       ratio: this.ratio,
 490       hasAudio: this.hasAudio,
 491
 492       segmentListSize: VIDEO_LIVE.SEGMENTS_LIST_SIZE,
 493       segmentDuration: getLiveSegmentTime(this.videoLive.latencyMode),
 494
 495       outDirectory: this.outDirectory
 496     }
 497
 498     return CONFIG.LIVE.TRANSCODING.ENABLED && CONFIG.LIVE.TRANSCODING.REMOTE_RUNNERS.ENABLED
 499       ? new RemoteTranscodingWrapper(options)
 500       : new FFmpegTranscodingWrapper(options)
 501   }
 502
 503   private getPlaylistIdFromTS (segmentPath: string) {
 504     const playlistIdMatcher = /^([\d+])-/
 505
 506     return basename(segmentPath).match(playlistIdMatcher)[1]
 507   }
 508
 509   private getPlaylistNameFromTS (segmentPath: string) {
 510     return `${this.getPlaylistIdFromTS(segmentPath)}.m3u8`
 511   }
 512 }
 513
 514 // ---------------------------------------------------------------------------
 515
 516 export {
 517   MuxingSession
 518 }