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 120')
  25     command.outputOption('-x264-params "no-scenecut=1"')
  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     command.on('stdout', data => console.log(data))
  43   }
  44
  45   command.run()
  46
  47   return command
  48 }
  49
  50 function waitFfmpegUntilError (command: FfmpegCommand, successAfterMS = 10000) {
  51   return new Promise<void>((res, rej) => {
  52     command.on('error', err => {
  53       return rej(err)
  54     })
  55
  56     setTimeout(() => {
  57       res()
  58     }, successAfterMS)
  59   })
  60 }
  61
  62 async function testFfmpegStreamError (command: FfmpegCommand, shouldHaveError: boolean) {
  63   let error: Error
  64
  65   try {
  66     await waitFfmpegUntilError(command, 45000)
  67   } catch (err) {
  68     error = err
  69   }
  70
  71   await stopFfmpeg(command)
  72
  73   if (shouldHaveError && !error) throw new Error('Ffmpeg did not have an error')
  74   if (!shouldHaveError && error) throw error
  75 }
  76
  77 async function stopFfmpeg (command: FfmpegCommand) {
  78   command.kill('SIGINT')
  79
  80   await wait(500)
  81 }
  82
  83 async function waitUntilLivePublishedOnAllServers (servers: PeerTubeServer[], videoId: string) {
  84   for (const server of servers) {
  85     await server.live.waitUntilPublished({ videoId })
  86   }
  87 }
  88
  89 async function waitUntilLiveWaitingOnAllServers (servers: PeerTubeServer[], videoId: string) {
  90   for (const server of servers) {
  91     await server.live.waitUntilWaiting({ videoId })
  92   }
  93 }
  94
  95 async function waitUntilLiveReplacedByReplayOnAllServers (servers: PeerTubeServer[], videoId: string) {
  96   for (const server of servers) {
  97     await server.live.waitUntilReplacedByReplay({ videoId })
  98   }
  99 }
 100
 101 async function findExternalSavedVideo (server: PeerTubeServer, liveDetails: VideoDetails) {
 102   const include = VideoInclude.BLACKLISTED
 103   const privacyOneOf = [ VideoPrivacy.INTERNAL, VideoPrivacy.PRIVATE, VideoPrivacy.PUBLIC, VideoPrivacy.UNLISTED ]
 104
 105   const { data } = await server.videos.list({ token: server.accessToken, sort: '-publishedAt', include, privacyOneOf })
 106
 107   return data.find(v => v.name === liveDetails.name + ' - ' + new Date(liveDetails.publishedAt).toLocaleString())
 108 }
 109
 110 export {
 111   sendRTMPStream,
 112   waitFfmpegUntilError,
 113   testFfmpegStreamError,
 114   stopFfmpeg,
 115
 116   waitUntilLivePublishedOnAllServers,
 117   waitUntilLiveReplacedByReplayOnAllServers,
 118   waitUntilLiveWaitingOnAllServers,
 119
 120   findExternalSavedVideo
 121 }