server/lib/runners/job-handlers/live-rtmp-hls-transcoding-job-handler.ts

   1 import { move, remove } from 'fs-extra'
   2 import { join } from 'path'
   3 import { logger } from '@server/helpers/logger'
   4 import { JOB_PRIORITY } from '@server/initializers/constants'
   5 import { LiveManager } from '@server/lib/live'
   6 import { MStreamingPlaylist, MVideo } from '@server/types/models'
   7 import { MRunnerJob } from '@server/types/models/runners'
   8 import { buildUUID } from '@shared/extra-utils'
   9 import {
  10   LiveRTMPHLSTranscodingSuccess,
  11   LiveRTMPHLSTranscodingUpdatePayload,
  12   LiveVideoError,
  13   RunnerJobLiveRTMPHLSTranscodingPayload,
  14   RunnerJobLiveRTMPHLSTranscodingPrivatePayload,
  15   RunnerJobState
  16 } from '@shared/models'
  17 import { AbstractJobHandler } from './abstract-job-handler'
  18
  19 type CreateOptions = {
  20   video: MVideo
  21   playlist: MStreamingPlaylist
  22
  23   rtmpUrl: string
  24
  25   toTranscode: {
  26     resolution: number
  27     fps: number
  28   }[]
  29
  30   segmentListSize: number
  31   segmentDuration: number
  32
  33   outputDirectory: string
  34 }
  35
  36 // eslint-disable-next-line max-len
  37 export class LiveRTMPHLSTranscodingJobHandler extends AbstractJobHandler<CreateOptions, LiveRTMPHLSTranscodingUpdatePayload, LiveRTMPHLSTranscodingSuccess> {
  38
  39   async create (options: CreateOptions) {
  40     const { video, rtmpUrl, toTranscode, playlist, segmentDuration, segmentListSize, outputDirectory } = options
  41
  42     const jobUUID = buildUUID()
  43     const payload: RunnerJobLiveRTMPHLSTranscodingPayload = {
  44       input: {
  45         rtmpUrl
  46       },
  47       output: {
  48         toTranscode,
  49         segmentListSize,
  50         segmentDuration
  51       }
  52     }
  53
  54     const privatePayload: RunnerJobLiveRTMPHLSTranscodingPrivatePayload = {
  55       videoUUID: video.uuid,
  56       masterPlaylistName: playlist.playlistFilename,
  57       outputDirectory
  58     }
  59
  60     const job = await this.createRunnerJob({
  61       type: 'live-rtmp-hls-transcoding',
  62       jobUUID,
  63       payload,
  64       privatePayload,
  65       priority: JOB_PRIORITY.TRANSCODING
  66     })
  67
  68     return job
  69   }
  70
  71   // ---------------------------------------------------------------------------
  72
  73   protected async specificUpdate (options: {
  74     runnerJob: MRunnerJob
  75     updatePayload: LiveRTMPHLSTranscodingUpdatePayload
  76   }) {
  77     const { runnerJob, updatePayload } = options
  78
  79     const privatePayload = runnerJob.privatePayload as RunnerJobLiveRTMPHLSTranscodingPrivatePayload
  80     const outputDirectory = privatePayload.outputDirectory
  81     const videoUUID = privatePayload.videoUUID
  82
  83     // Always process the chunk first before moving m3u8 that references this chunk
  84     if (updatePayload.type === 'add-chunk') {
  85       await move(
  86         updatePayload.videoChunkFile as string,
  87         join(outputDirectory, updatePayload.videoChunkFilename),
  88         { overwrite: true }
  89       )
  90     } else if (updatePayload.type === 'remove-chunk') {
  91       await remove(join(outputDirectory, updatePayload.videoChunkFilename))
  92     }
  93
  94     if (updatePayload.resolutionPlaylistFile && updatePayload.resolutionPlaylistFilename) {
  95       await move(
  96         updatePayload.resolutionPlaylistFile as string,
  97         join(outputDirectory, updatePayload.resolutionPlaylistFilename),
  98         { overwrite: true }
  99       )
 100     }
 101
 102     if (updatePayload.masterPlaylistFile) {
 103       await move(updatePayload.masterPlaylistFile as string, join(outputDirectory, privatePayload.masterPlaylistName), { overwrite: true })
 104     }
 105
 106     logger.info(
 107       'Runner live RTMP to HLS job %s for %s updated.',
 108       runnerJob.uuid, videoUUID, { updatePayload, ...this.lTags(videoUUID, runnerJob.uuid) }
 109     )
 110   }
 111
 112   // ---------------------------------------------------------------------------
 113
 114   protected specificComplete (options: {
 115     runnerJob: MRunnerJob
 116   }) {
 117     return this.stopLive({
 118       runnerJob: options.runnerJob,
 119       type: 'ended'
 120     })
 121   }
 122
 123   // ---------------------------------------------------------------------------
 124
 125   protected isAbortSupported () {
 126     return false
 127   }
 128
 129   protected specificAbort () {
 130     throw new Error('Not implemented')
 131   }
 132
 133   protected specificError (options: {
 134     runnerJob: MRunnerJob
 135     nextState: RunnerJobState
 136   }) {
 137     return this.stopLive({
 138       runnerJob: options.runnerJob,
 139       type: 'errored'
 140     })
 141   }
 142
 143   protected specificCancel (options: {
 144     runnerJob: MRunnerJob
 145   }) {
 146     return this.stopLive({
 147       runnerJob: options.runnerJob,
 148       type: 'cancelled'
 149     })
 150   }
 151
 152   private stopLive (options: {
 153     runnerJob: MRunnerJob
 154     type: 'ended' | 'errored' | 'cancelled'
 155   }) {
 156     const { runnerJob, type } = options
 157
 158     const privatePayload = runnerJob.privatePayload as RunnerJobLiveRTMPHLSTranscodingPrivatePayload
 159     const videoUUID = privatePayload.videoUUID
 160
 161     const errorType = {
 162       ended: null,
 163       errored: LiveVideoError.RUNNER_JOB_ERROR,
 164       cancelled: LiveVideoError.RUNNER_JOB_CANCEL
 165     }
 166
 167     LiveManager.Instance.stopSessionOf(privatePayload.videoUUID, errorType[type])
 168
 169     logger.info('Runner live RTMP to HLS job %s for video %s %s.', runnerJob.uuid, videoUUID, type, this.lTags(runnerJob.uuid, videoUUID))
 170   }
 171 }