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 { 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
  11 export type JobWithToken <T extends RunnerJobPayload = RunnerJobPayload> = RunnerJob<T> & { jobToken: string }
  12
  13 export type ProcessOptions <T extends RunnerJobPayload = RunnerJobPayload> = {
  14   server: PeerTubeServer
  15   job: JobWithToken<T>
  16   runnerToken: string
  17 }
  18
  19 export async function downloadInputFile (options: {
  20   url: string
  21   job: JobWithToken
  22   runnerToken: string
  23 }) {
  24   const { url, job, runnerToken } = options
  25   const destination = join(ConfigManager.Instance.getTranscodingDirectory(), buildUUID())
  26
  27   await downloadFile({ url, jobToken: job.jobToken, runnerToken, destination })
  28
  29   return destination
  30 }
  31
  32 export async function updateTranscodingProgress (options: {
  33   server: PeerTubeServer
  34   runnerToken: string
  35   job: JobWithToken
  36   progress: number
  37 }) {
  38   const { server, job, runnerToken, progress } = options
  39
  40   return server.runnerJobs.update({ jobToken: job.jobToken, jobUUID: job.uuid, runnerToken, progress })
  41 }
  42
  43 export function buildFFmpegVOD (options: {
  44   server: PeerTubeServer
  45   runnerToken: string
  46   job: JobWithToken
  47 }) {
  48   const { server, job, runnerToken } = options
  49
  50   const updateInterval = ConfigManager.Instance.isTestInstance()
  51     ? 500
  52     : 60000
  53
  54   const updateJobProgress = throttle((progress: number) => {
  55     if (progress < 0 || progress > 100) progress = undefined
  56
  57     updateTranscodingProgress({ server, job, runnerToken, progress })
  58       .catch(err => logger.error({ err }, 'Cannot send job progress'))
  59   }, updateInterval, { trailing: false })
  60
  61   const config = ConfigManager.Instance.getConfig()
  62
  63   return new FFmpegVOD({
  64     niceness: config.ffmpeg.nice,
  65     threads: config.ffmpeg.threads,
  66     tmpDirectory: ConfigManager.Instance.getTranscodingDirectory(),
  67     profile: 'default',
  68     availableEncoders: {
  69       available: getAvailableEncoders(),
  70       encodersToTry: getEncodersToTry()
  71     },
  72     logger: getTranscodingLogger(),
  73     updateJobProgress
  74   })
  75 }
  76
  77 export function buildFFmpegLive () {
  78   const config = ConfigManager.Instance.getConfig()
  79
  80   return new FFmpegLive({
  81     niceness: config.ffmpeg.nice,
  82     threads: config.ffmpeg.threads,
  83     tmpDirectory: ConfigManager.Instance.getTranscodingDirectory(),
  84     profile: 'default',
  85     availableEncoders: {
  86       available: getAvailableEncoders(),
  87       encodersToTry: getEncodersToTry()
  88     },
  89     logger: getTranscodingLogger()
  90   })
  91 }