shared/server-commands/videos/live.ts

   1 import ffmpeg, { FfmpegCommand } from 'fluent-ffmpeg'
   2 import { buildAbsoluteFixturePath, wait } from '@shared/core-utils'
   3 import { VideoDetails, VideoInclude, VideoPrivacy } from '@shared/models'
   4 import { PeerTubeServer } from '../server/server'
   5
   6 function sendRTMPStream (options: {
   7   rtmpBaseUrl: string
   8   streamKey: string
   9   fixtureName?: string // default video_short.mp4
  10   copyCodecs?: boolean // default false
  11 }) {
  12   const { rtmpBaseUrl, streamKey, fixtureName = 'video_short.mp4', copyCodecs = false } = options
  13
  14   const fixture = buildAbsoluteFixturePath(fixtureName)
  15
  16   const command = ffmpeg(fixture)
  17   command.inputOption('-stream_loop -1')
  18   command.inputOption('-re')
  19
  20   if (copyCodecs) {
  21     command.outputOption('-c copy')
  22   } else {
  23     command.outputOption('-c:v libx264')
  24     command.outputOption('-g 50')
  25     command.outputOption('-keyint_min 2')
  26     command.outputOption('-r 60')
  27   }
  28
  29   command.outputOption('-f flv')
  30
  31   const rtmpUrl = rtmpBaseUrl + '/' + streamKey
  32   command.output(rtmpUrl)
  33
  34   command.on('error', err => {
  35     if (err?.message?.includes('Exiting normally')) return
  36
  37     if (process.env.DEBUG) console.error(err)
  38   })
  39
  40   if (process.env.DEBUG) {
  41     command.on('stderr', data => console.log(data))
  42   }
  43
  44   command.run()
  45
  46   return command
  47 }
  48
  49 function waitFfmpegUntilError (command: FfmpegCommand, successAfterMS = 10000) {
  50   return new Promise<void>((res, rej) => {
  51     command.on('error', err => {
  52       return rej(err)
  53     })
  54
  55     setTimeout(() => {
  56       res()
  57     }, successAfterMS)
  58   })
  59 }
  60
  61 async function testFfmpegStreamError (command: FfmpegCommand, shouldHaveError: boolean) {
  62   let error: Error
  63
  64   try {
  65     await waitFfmpegUntilError(command, 35000)
  66   } catch (err) {
  67     error = err
  68   }
  69
  70   await stopFfmpeg(command)
  71
  72   if (shouldHaveError && !error) throw new Error('Ffmpeg did not have an error')
  73   if (!shouldHaveError && error) throw error
  74 }
  75
  76 async function stopFfmpeg (command: FfmpegCommand) {
  77   command.kill('SIGINT')
  78
  79   await wait(500)
  80 }
  81
  82 async function waitUntilLivePublishedOnAllServers (servers: PeerTubeServer[], videoId: string) {
  83   for (const server of servers) {
  84     await server.live.waitUntilPublished({ videoId })
  85   }
  86 }
  87
  88 async function waitUntilLiveWaitingOnAllServers (servers: PeerTubeServer[], videoId: string) {
  89   for (const server of servers) {
  90     await server.live.waitUntilWaiting({ videoId })
  91   }
  92 }
  93
  94 async function waitUntilLiveReplacedByReplayOnAllServers (servers: PeerTubeServer[], videoId: string) {
  95   for (const server of servers) {
  96     await server.live.waitUntilReplacedByReplay({ videoId })
  97   }
  98 }
  99
 100 async function findExternalSavedVideo (server: PeerTubeServer, liveDetails: VideoDetails) {
 101   const include = VideoInclude.BLACKLISTED
 102   const privacyOneOf = [ VideoPrivacy.INTERNAL, VideoPrivacy.PRIVATE, VideoPrivacy.PUBLIC, VideoPrivacy.UNLISTED ]
 103
 104   const { data } = await server.videos.list({ token: server.accessToken, sort: '-publishedAt', include, privacyOneOf })
 105
 106   return data.find(v => v.name === liveDetails.name + ' - ' + new Date(liveDetails.publishedAt).toLocaleString())
 107 }
 108
 109 export {
 110   sendRTMPStream,
 111   waitFfmpegUntilError,
 112   testFfmpegStreamError,
 113   stopFfmpeg,
 114
 115   waitUntilLivePublishedOnAllServers,
 116   waitUntilLiveReplacedByReplayOnAllServers,
 117   waitUntilLiveWaitingOnAllServers,
 118
 119   findExternalSavedVideo
 120 }