Support progress for ffmpeg tasks

[github/Chocobozzz/PeerTube.git] / server / lib / live-manager.ts

diff --git a/server/lib/live-manager.ts b/server/lib/live-manager.ts
index dcf0161694afaf6750a432bf103c388d89f7fe79..7f5fdf8992835af7038f41dd8c93accc0ab8fc77 100644 (file)
--- a/server/lib/live-manager.ts
+++ b/server/lib/live-manager.ts
@@ -1,7 +1,8 @@
 
+import * as Bluebird from 'bluebird'
 import * as chokidar from 'chokidar'
 import { FfmpegCommand } from 'fluent-ffmpeg'
-import { copy, ensureDir, stat } from 'fs-extra'
+import { appendFile, ensureDir, readFile, stat } from 'fs-extra'
 import { basename, join } from 'path'
 import { isTestInstance } from '@server/helpers/core-utils'
 import { getLiveMuxingCommand, getLiveTranscodingCommand } from '@server/helpers/ffmpeg-utils'
@@ -108,6 +109,10 @@ class LiveManager {
     logger.info('Running RTMP server on port %d', config.rtmp.port)
 
     this.rtmpServer = new NodeRtmpServer(config)
+    this.rtmpServer.tcpServer.on('error', err => {
+      logger.error('Cannot run RTMP server.', { err })
+    })
+
     this.rtmpServer.run()
   }
 
@@ -158,6 +163,32 @@ class LiveManager {
     this.segmentsSha256.delete(videoUUID)
   }
 
+  addSegmentToReplay (hlsVideoPath: string, segmentPath: string) {
+    const segmentName = basename(segmentPath)
+    const dest = join(hlsVideoPath, VIDEO_LIVE.REPLAY_DIRECTORY, this.buildConcatenatedName(segmentName))
+
+    return readFile(segmentPath)
+      .then(data => appendFile(dest, data))
+      .catch(err => logger.error('Cannot copy segment %s to repay directory.', segmentPath, { err }))
+  }
+
+  buildConcatenatedName (segmentOrPlaylistPath: string) {
+    const num = basename(segmentOrPlaylistPath).match(/^(\d+)(-|\.)/)
+
+    return 'concat-' + num[1] + '.ts'
+  }
+
+  private processSegments (hlsVideoPath: string, videoUUID: string, videoLive: MVideoLive, segmentPaths: string[]) {
+    Bluebird.mapSeries(segmentPaths, async previousSegment => {
+      // Add sha hash of previous segments, because ffmpeg should have finished generating them
+      await this.addSegmentSha(videoUUID, previousSegment)
+
+      if (videoLive.saveReplay) {
+        await this.addSegmentToReplay(hlsVideoPath, previousSegment)
+      }
+    }).catch(err => logger.error('Cannot process segments in %s', hlsVideoPath, { err }))
+  }
+
   private getContext () {
     return context
   }
@@ -260,16 +291,17 @@ class LiveManager {
     for (let i = 0; i < allResolutions.length; i++) {
       const resolution = allResolutions[i]
 
-      VideoFileModel.upsert({
+      const file = new VideoFileModel({
         resolution,
         size: -1,
         extname: '.ts',
         infoHash: null,
         fps,
         videoStreamingPlaylistId: playlist.id
-      }).catch(err => {
-        logger.error('Cannot create file for live streaming.', { err })
       })
+
+      VideoFileModel.customUpsert(file, 'streaming-playlist', null)
+        .catch(err => logger.error('Cannot create file for live streaming.', { err }))
     }
 
     const outPath = getHLSDirectory(videoLive.Video)
@@ -302,28 +334,13 @@ class LiveManager {
     const segmentsToProcessPerPlaylist: { [playlistId: string]: string[] } = {}
     const playlistIdMatcher = /^([\d+])-/
 
-    const processSegments = (segmentsToProcess: string[]) => {
-      // 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 }))
-
-        if (videoLive.saveReplay) {
-          const segmentName = basename(previousSegment)
-
-          copy(previousSegment, join(outPath, VIDEO_LIVE.REPLAY_DIRECTORY, segmentName))
-            .catch(err => logger.error('Cannot copy segment %s to repay directory.', previousSegment, { err }))
-        }
-      }
-    }
-
     const addHandler = segmentPath => {
       logger.debug('Live add handler of %s.', segmentPath)
 
       const playlistId = basename(segmentPath).match(playlistIdMatcher)[0]
 
       const segmentsToProcess = segmentsToProcessPerPlaylist[playlistId] || []
-      processSegments(segmentsToProcess)
+      this.processSegments(outPath, videoUUID, videoLive, segmentsToProcess)
 
       segmentsToProcessPerPlaylist[playlistId] = [ segmentPath ]
 
@@ -400,7 +417,7 @@ class LiveManager {
           .then(() => {
             // Process remaining segments hash
             for (const key of Object.keys(segmentsToProcessPerPlaylist)) {
-              processSegments(segmentsToProcessPerPlaylist[key])
+              this.processSegments(outPath, videoUUID, videoLive, segmentsToProcessPerPlaylist[key])
             }
           })
           .catch(err => logger.error('Cannot close watchers of %s or process remaining hash segments.', outPath, { err }))
@@ -492,7 +509,7 @@ class LiveManager {
   private isDurationConstraintValid (streamingStartTime: number) {
     const maxDuration = CONFIG.LIVE.MAX_DURATION
     // No limit
-    if (maxDuration === null) return true
+    if (maxDuration < 0) return true
 
     const now = new Date().getTime()
     const max = streamingStartTime + maxDuration
@@ -524,6 +541,8 @@ class LiveManager {
 
       await federateVideoIfNeeded(video, false)
 
+      PeerTubeSocket.Instance.sendVideoViewsUpdate(video)
+
       // Only keep not expired watchers
       const newWatchers = watchers.filter(w => w > notBefore)
       this.watchersPerVideo.set(videoId, newWatchers)