X-Git-Url: https://git.immae.eu/?a=blobdiff_plain;f=server%2Flib%2Flive-manager.ts;h=60ef30d15641cd5478defb4a0fdab65f018ba5ea;hb=2122606318bf208262d1573ab8d9a772cd697448;hp=e85998686ff41365c68a844b2a1f727ffed7b02b;hpb=c655c9ef6f7ddb47a153adc04d5c10c6de3d5ed7;p=github%2FChocobozzz%2FPeerTube.git

diff --git a/server/lib/live-manager.ts b/server/lib/live-manager.ts
index e85998686..60ef30d15 100644
--- a/server/lib/live-manager.ts
+++ b/server/lib/live-manager.ts
@@ -1,5 +1,4 @@
 
-import { AsyncQueue, queue } from 'async'
 import * as chokidar from 'chokidar'
 import { FfmpegCommand } from 'fluent-ffmpeg'
 import { ensureDir, stat } from 'fs-extra'
@@ -50,12 +49,6 @@ const config = {
   }
 }
 
-type SegmentSha256QueueParam = {
-  operation: 'update' | 'delete'
-  videoUUID: string
-  segmentPath: string
-}
-
 class LiveManager {
 
   private static instance: LiveManager
@@ -71,7 +64,6 @@ class LiveManager {
     return isAbleToUploadVideo(userId, 1000)
   }, { maxAge: MEMOIZE_TTL.LIVE_ABLE_TO_UPLOAD })
 
-  private segmentsSha256Queue: AsyncQueue<SegmentSha256QueueParam>
   private rtmpServer: any
 
   private constructor () {
@@ -96,18 +88,6 @@ class LiveManager {
       logger.info('Live session ended.', { sessionId })
     })
 
-    this.segmentsSha256Queue = queue<SegmentSha256QueueParam, Error>((options, cb) => {
-      const promise = options.operation === 'update'
-        ? this.addSegmentSha(options)
-        : Promise.resolve(this.removeSegmentSha(options))
-
-      promise.then(() => cb())
-        .catch(err => {
-          logger.error('Cannot update/remove sha segment %s.', options.segmentPath, { err })
-          cb()
-        })
-    })
-
     registerConfigChangedHandler(() => {
       if (!this.rtmpServer && CONFIG.LIVE.ENABLED === true) {
         this.run()
@@ -219,13 +199,15 @@ class LiveManager {
       ? computeResolutionsToTranscode(resolutionResult.videoFileResolution, 'live')
       : []
 
-    logger.info('Will mux/transcode live video of original resolution %d.', session.videoHeight, { resolutionsEnabled })
+    const allResolutions = resolutionsEnabled.concat([ session.videoHeight ])
+
+    logger.info('Will mux/transcode live video of original resolution %d.', session.videoHeight, { allResolutions })
 
     const [ videoStreamingPlaylist ] = await VideoStreamingPlaylistModel.upsert({
       videoId: video.id,
       playlistUrl,
       segmentsSha256Url: WEBSERVER.URL + VideoStreamingPlaylistModel.getHlsSha256SegmentsStaticPath(video.uuid, video.isLive),
-      p2pMediaLoaderInfohashes: VideoStreamingPlaylistModel.buildP2PMediaLoaderInfoHashes(playlistUrl, resolutionsEnabled),
+      p2pMediaLoaderInfohashes: VideoStreamingPlaylistModel.buildP2PMediaLoaderInfoHashes(playlistUrl, allResolutions),
       p2pMediaLoaderPeerVersion: P2P_MEDIA_LOADER_PEER_VERSION,
 
       type: VideoStreamingPlaylistType.HLS
@@ -235,10 +217,9 @@ class LiveManager {
       sessionId,
       videoLive,
       playlist: videoStreamingPlaylist,
-      originalResolution: session.videoHeight,
       rtmpUrl,
       fps,
-      resolutionsEnabled
+      allResolutions
     })
   }
 
@@ -248,12 +229,10 @@ class LiveManager {
     playlist: MStreamingPlaylist
     rtmpUrl: string
     fps: number
-    resolutionsEnabled: number[]
-    originalResolution: number
+    allResolutions: number[]
   }) {
-    const { sessionId, videoLive, playlist, resolutionsEnabled, originalResolution, fps, rtmpUrl } = options
+    const { sessionId, videoLive, playlist, allResolutions, fps, rtmpUrl } = options
     const startStreamDateTime = new Date().getTime()
-    const allResolutions = resolutionsEnabled.concat([ originalResolution ])
 
     const user = await UserModel.loadByLiveId(videoLive.id)
     if (!this.livesPerUser.has(user.id)) {
@@ -294,11 +273,18 @@ class LiveManager {
 
     const tsWatcher = chokidar.watch(outPath + '/*.ts')
 
-    const updateSegment = segmentPath => this.segmentsSha256Queue.push({ operation: 'update', segmentPath, videoUUID })
+    let segmentsToProcess: string[] = []
 
     const addHandler = segmentPath => {
-      updateSegment(segmentPath)
+      // Add sha hash of previous segments, because ffmpeg should have finished generating them
+      for (const previousSegment of segmentsToProcess) {
+        this.addSegmentSha(videoUUID, previousSegment)
+          .catch(err => logger.error('Cannot add sha segment of video %s -> %s.', videoUUID, previousSegment, { err }))
+      }
+
+      segmentsToProcess = [ segmentPath ]
 
+      // Duration constraint check
       if (this.isDurationConstraintValid(startStreamDateTime) !== true) {
         logger.info('Stopping session of %s: max duration exceeded.', videoUUID)
 
@@ -323,10 +309,9 @@ class LiveManager {
       }
     }
 
-    const deleteHandler = segmentPath => this.segmentsSha256Queue.push({ operation: 'delete', segmentPath, videoUUID })
+    const deleteHandler = segmentPath => this.removeSegmentSha(videoUUID, segmentPath)
 
     tsWatcher.on('add', p => addHandler(p))
-    tsWatcher.on('change', p => updateSegment(p))
     tsWatcher.on('unlink', p => deleteHandler(p))
 
     const masterWatcher = chokidar.watch(outPath + '/master.m3u8')
@@ -399,33 +384,33 @@ class LiveManager {
     }
   }
 
-  private async addSegmentSha (options: SegmentSha256QueueParam) {
-    const segmentName = basename(options.segmentPath)
-    logger.debug('Updating live sha segment %s.', options.segmentPath)
+  private async addSegmentSha (videoUUID: string, segmentPath: string) {
+    const segmentName = basename(segmentPath)
+    logger.debug('Adding live sha segment %s.', segmentPath)
 
-    const shaResult = await buildSha256Segment(options.segmentPath)
+    const shaResult = await buildSha256Segment(segmentPath)
 
-    if (!this.segmentsSha256.has(options.videoUUID)) {
-      this.segmentsSha256.set(options.videoUUID, new Map())
+    if (!this.segmentsSha256.has(videoUUID)) {
+      this.segmentsSha256.set(videoUUID, new Map())
     }
 
-    const filesMap = this.segmentsSha256.get(options.videoUUID)
+    const filesMap = this.segmentsSha256.get(videoUUID)
     filesMap.set(segmentName, shaResult)
   }
 
-  private removeSegmentSha (options: SegmentSha256QueueParam) {
-    const segmentName = basename(options.segmentPath)
+  private removeSegmentSha (videoUUID: string, segmentPath: string) {
+    const segmentName = basename(segmentPath)
 
-    logger.debug('Removing live sha segment %s.', options.segmentPath)
+    logger.debug('Removing live sha segment %s.', segmentPath)
 
-    const filesMap = this.segmentsSha256.get(options.videoUUID)
+    const filesMap = this.segmentsSha256.get(videoUUID)
     if (!filesMap) {
-      logger.warn('Unknown files map to remove sha for %s.', options.videoUUID)
+      logger.warn('Unknown files map to remove sha for %s.', videoUUID)
       return
     }
 
     if (!filesMap.has(segmentName)) {
-      logger.warn('Unknown segment in files map for video %s and segment %s.', options.videoUUID, options.segmentPath)
+      logger.warn('Unknown segment in files map for video %s and segment %s.', videoUUID, segmentPath)
       return
     }