1 files changed, 100 insertions, 0 deletions

diff --git a/shared/server-commands/videos/live.ts b/shared/server-commands/videos/live.ts
new file mode 100644
index 000000000..7a7faa911
--- /dev/null
+++ b/shared/server-commands/videos/live.ts
@@ -0,0 +1,100 @@
+import ffmpeg, { FfmpegCommand } from 'fluent-ffmpeg'
+import { buildAbsoluteFixturePath, wait } from '@shared/core-utils'
+import { PeerTubeServer } from '../server/server'
+
+function sendRTMPStream (options: {
+  rtmpBaseUrl: string
+  streamKey: string
+  fixtureName?: string // default video_short.mp4
+  copyCodecs?: boolean // default false
+}) {
+  const { rtmpBaseUrl, streamKey, fixtureName = 'video_short.mp4', copyCodecs = false } = options
+
+  const fixture = buildAbsoluteFixturePath(fixtureName)
+
+  const command = ffmpeg(fixture)
+  command.inputOption('-stream_loop -1')
+  command.inputOption('-re')
+
+  if (copyCodecs) {
+    command.outputOption('-c copy')
+  } else {
+    command.outputOption('-c:v libx264')
+    command.outputOption('-g 50')
+    command.outputOption('-keyint_min 2')
+    command.outputOption('-r 60')
+  }
+
+  command.outputOption('-f flv')
+
+  const rtmpUrl = rtmpBaseUrl + '/' + streamKey
+  command.output(rtmpUrl)
+
+  command.on('error', err => {
+    if (err?.message?.includes('Exiting normally')) return
+
+    if (process.env.DEBUG) console.error(err)
+  })
+
+  if (process.env.DEBUG) {
+    command.on('stderr', data => console.log(data))
+  }
+
+  command.run()
+
+  return command
+}
+
+function waitFfmpegUntilError (command: FfmpegCommand, successAfterMS = 10000) {
+  return new Promise<void>((res, rej) => {
+    command.on('error', err => {
+      return rej(err)
+    })
+
+    setTimeout(() => {
+      res()
+    }, successAfterMS)
+  })
+}
+
+async function testFfmpegStreamError (command: FfmpegCommand, shouldHaveError: boolean) {
+  let error: Error
+
+  try {
+    await waitFfmpegUntilError(command, 35000)
+  } catch (err) {
+    error = err
+  }
+
+  await stopFfmpeg(command)
+
+  if (shouldHaveError && !error) throw new Error('Ffmpeg did not have an error')
+  if (!shouldHaveError && error) throw error
+}
+
+async function stopFfmpeg (command: FfmpegCommand) {
+  command.kill('SIGINT')
+
+  await wait(500)
+}
+
+async function waitUntilLivePublishedOnAllServers (servers: PeerTubeServer[], videoId: string) {
+  for (const server of servers) {
+    await server.live.waitUntilPublished({ videoId })
+  }
+}
+
+async function waitUntilLiveSavedOnAllServers (servers: PeerTubeServer[], videoId: string) {
+  for (const server of servers) {
+    await server.live.waitUntilSaved({ videoId })
+  }
+}
+
+export {
+  sendRTMPStream,
+  waitFfmpegUntilError,
+  testFfmpegStreamError,
+  stopFfmpeg,
+  waitUntilLivePublishedOnAllServers,
+  waitUntilLiveSavedOnAllServers
+}