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