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