packages/peertube-runner/server/process/shared/common.ts

   1 import { throttle } from 'lodash'
   2 import { ConfigManager, downloadFile, logger } from 'packages/peertube-runner/shared'
   3 import { join } from 'path'
   4 import { buildUUID } from '@shared/extra-utils'
   5 import { FFmpegEdition, FFmpegLive, FFmpegVOD } from '@shared/ffmpeg'
   6 import { RunnerJob, RunnerJobPayload } from '@shared/models'
   7 import { PeerTubeServer } from '@shared/server-commands'
   8 import { getTranscodingLogger } from './transcoding-logger'
   9 import { getAvailableEncoders, getEncodersToTry } from './transcoding-profiles'
  10 import { remove } from 'fs-extra'
  11
  12 export type JobWithToken <T extends RunnerJobPayload = RunnerJobPayload> = RunnerJob<T> & { jobToken: string }
  13
  14 export type ProcessOptions <T extends RunnerJobPayload = RunnerJobPayload> = {
  15   server: PeerTubeServer
  16   job: JobWithToken<T>
  17   runnerToken: string
  18 }
  19
  20 export async function downloadInputFile (options: {
  21   url: string
  22   job: JobWithToken
  23   runnerToken: string
  24 }) {
  25   const { url, job, runnerToken } = options
  26   const destination = join(ConfigManager.Instance.getTranscodingDirectory(), buildUUID())
  27
  28   try {
  29     await downloadFile({ url, jobToken: job.jobToken, runnerToken, destination })
  30   } catch (err) {
  31     remove(destination)
  32       .catch(err => logger.error({ err }, `Cannot remove ${destination}`))
  33
  34     throw err
  35   }
  36
  37   return destination
  38 }
  39
  40 export async function updateTranscodingProgress (options: {
  41   server: PeerTubeServer
  42   runnerToken: string
  43   job: JobWithToken
  44   progress: number
  45 }) {
  46   const { server, job, runnerToken, progress } = options
  47
  48   return server.runnerJobs.update({ jobToken: job.jobToken, jobUUID: job.uuid, runnerToken, progress })
  49 }
  50
  51 // ---------------------------------------------------------------------------
  52
  53 export function buildFFmpegVOD (options: {
  54   server: PeerTubeServer
  55   runnerToken: string
  56   job: JobWithToken
  57 }) {
  58   const { server, job, runnerToken } = options
  59
  60   const updateInterval = ConfigManager.Instance.isTestInstance()
  61     ? 500
  62     : 60000
  63
  64   const updateJobProgress = throttle((progress: number) => {
  65     if (progress < 0 || progress > 100) progress = undefined
  66
  67     updateTranscodingProgress({ server, job, runnerToken, progress })
  68       .catch(err => logger.error({ err }, 'Cannot send job progress'))
  69   }, updateInterval, { trailing: false })
  70
  71   return new FFmpegVOD({
  72     ...getCommonFFmpegOptions(),
  73
  74     updateJobProgress
  75   })
  76 }
  77
  78 export function buildFFmpegLive () {
  79   return new FFmpegLive(getCommonFFmpegOptions())
  80 }
  81
  82 export function buildFFmpegEdition () {
  83   return new FFmpegEdition(getCommonFFmpegOptions())
  84 }
  85
  86 function getCommonFFmpegOptions () {
  87   const config = ConfigManager.Instance.getConfig()
  88
  89   return {
  90     niceness: config.ffmpeg.nice,
  91     threads: config.ffmpeg.threads,
  92     tmpDirectory: ConfigManager.Instance.getTranscodingDirectory(),
  93     profile: 'default',
  94     availableEncoders: {
  95       available: getAvailableEncoders(),
  96       encodersToTry: getEncodersToTry()
  97     },
  98     logger: getTranscodingLogger()
  99   }
 100 }