From 0c9668f77901e7540e2c7045eb0f2974a4842a69 Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Fri, 21 Apr 2023 14:55:10 +0200 Subject: Implement remote runner jobs in server Move ffmpeg functions to @shared --- .../abstract-transcoding-wrapper.ts | 101 +++++++++++++++++++++ .../ffmpeg-transcoding-wrapper.ts | 95 +++++++++++++++++++ .../lib/live/shared/transcoding-wrapper/index.ts | 3 + .../remote-transcoding-wrapper.ts | 20 ++++ 4 files changed, 219 insertions(+) create mode 100644 server/lib/live/shared/transcoding-wrapper/abstract-transcoding-wrapper.ts create mode 100644 server/lib/live/shared/transcoding-wrapper/ffmpeg-transcoding-wrapper.ts create mode 100644 server/lib/live/shared/transcoding-wrapper/index.ts create mode 100644 server/lib/live/shared/transcoding-wrapper/remote-transcoding-wrapper.ts (limited to 'server/lib/live/shared/transcoding-wrapper') diff --git a/server/lib/live/shared/transcoding-wrapper/abstract-transcoding-wrapper.ts b/server/lib/live/shared/transcoding-wrapper/abstract-transcoding-wrapper.ts new file mode 100644 index 000000000..226ba4573 --- /dev/null +++ b/server/lib/live/shared/transcoding-wrapper/abstract-transcoding-wrapper.ts @@ -0,0 +1,101 @@ +import EventEmitter from 'events' +import { LoggerTagsFn } from '@server/helpers/logger' +import { MStreamingPlaylistVideo, MVideoLiveVideo } from '@server/types/models' +import { LiveVideoError } from '@shared/models' + +interface TranscodingWrapperEvents { + 'end': () => void + + 'error': (options: { err: Error }) => void +} + +declare interface AbstractTranscodingWrapper { + on( + event: U, listener: TranscodingWrapperEvents[U] + ): this + + emit( + event: U, ...args: Parameters + ): boolean +} + +interface AbstractTranscodingWrapperOptions { + streamingPlaylist: MStreamingPlaylistVideo + videoLive: MVideoLiveVideo + + lTags: LoggerTagsFn + + inputUrl: string + fps: number + toTranscode: { + resolution: number + fps: number + }[] + + bitrate: number + ratio: number + hasAudio: boolean + + segmentListSize: number + segmentDuration: number + + outDirectory: string +} + +abstract class AbstractTranscodingWrapper extends EventEmitter { + protected readonly videoLive: MVideoLiveVideo + + protected readonly toTranscode: { + resolution: number + fps: number + }[] + + protected readonly inputUrl: string + protected readonly fps: number + protected readonly bitrate: number + protected readonly ratio: number + protected readonly hasAudio: boolean + + protected readonly segmentListSize: number + protected readonly segmentDuration: number + + protected readonly videoUUID: string + + protected readonly outDirectory: string + + protected readonly lTags: LoggerTagsFn + + protected readonly streamingPlaylist: MStreamingPlaylistVideo + + constructor (options: AbstractTranscodingWrapperOptions) { + super() + + this.lTags = options.lTags + + this.videoLive = options.videoLive + this.videoUUID = options.videoLive.Video.uuid + this.streamingPlaylist = options.streamingPlaylist + + this.inputUrl = options.inputUrl + this.fps = options.fps + this.toTranscode = options.toTranscode + + this.bitrate = options.bitrate + this.ratio = options.ratio + this.hasAudio = options.hasAudio + + this.segmentListSize = options.segmentListSize + this.segmentDuration = options.segmentDuration + + this.outDirectory = options.outDirectory + } + + abstract run (): Promise + + abstract abort (error?: LiveVideoError): void +} + +export { + AbstractTranscodingWrapper, + AbstractTranscodingWrapperOptions +} diff --git a/server/lib/live/shared/transcoding-wrapper/ffmpeg-transcoding-wrapper.ts b/server/lib/live/shared/transcoding-wrapper/ffmpeg-transcoding-wrapper.ts new file mode 100644 index 000000000..1f4c12bd4 --- /dev/null +++ b/server/lib/live/shared/transcoding-wrapper/ffmpeg-transcoding-wrapper.ts @@ -0,0 +1,95 @@ +import { FfmpegCommand } from 'fluent-ffmpeg' +import { getFFmpegCommandWrapperOptions } from '@server/helpers/ffmpeg' +import { logger } from '@server/helpers/logger' +import { CONFIG } from '@server/initializers/config' +import { VIDEO_LIVE } from '@server/initializers/constants' +import { VideoTranscodingProfilesManager } from '@server/lib/transcoding/default-transcoding-profiles' +import { FFmpegLive } from '@shared/ffmpeg' +import { getLiveSegmentTime } from '../../live-utils' +import { AbstractTranscodingWrapper } from './abstract-transcoding-wrapper' + +export class FFmpegTranscodingWrapper extends AbstractTranscodingWrapper { + private ffmpegCommand: FfmpegCommand + private ended = false + + async run () { + this.ffmpegCommand = CONFIG.LIVE.TRANSCODING.ENABLED + ? await this.buildFFmpegLive().getLiveTranscodingCommand({ + inputUrl: this.inputUrl, + + outPath: this.outDirectory, + masterPlaylistName: this.streamingPlaylist.playlistFilename, + + segmentListSize: this.segmentListSize, + segmentDuration: this.segmentDuration, + + toTranscode: this.toTranscode, + + bitrate: this.bitrate, + ratio: this.ratio, + + hasAudio: this.hasAudio + }) + : this.buildFFmpegLive().getLiveMuxingCommand({ + inputUrl: this.inputUrl, + outPath: this.outDirectory, + + masterPlaylistName: this.streamingPlaylist.playlistFilename, + + segmentListSize: VIDEO_LIVE.SEGMENTS_LIST_SIZE, + segmentDuration: getLiveSegmentTime(this.videoLive.latencyMode) + }) + + logger.info('Running local live muxing/transcoding for %s.', this.videoUUID, this.lTags()) + + this.ffmpegCommand.run() + + let ffmpegShellCommand: string + this.ffmpegCommand.on('start', cmdline => { + ffmpegShellCommand = cmdline + + logger.debug('Running ffmpeg command for live', { ffmpegShellCommand, ...this.lTags() }) + }) + + this.ffmpegCommand.on('error', (err, stdout, stderr) => { + this.onFFmpegError({ err, stdout, stderr, ffmpegShellCommand }) + }) + + this.ffmpegCommand.on('end', () => { + this.onFFmpegEnded() + }) + + this.ffmpegCommand.run() + } + + abort () { + // Nothing to do, ffmpeg will automatically exit + } + + private onFFmpegError (options: { + err: any + stdout: string + stderr: string + ffmpegShellCommand: string + }) { + const { err, stdout, stderr, ffmpegShellCommand } = options + + // Don't care that we killed the ffmpeg process + if (err?.message?.includes('Exiting normally')) return + + logger.error('FFmpeg transcoding error.', { err, stdout, stderr, ffmpegShellCommand, ...this.lTags() }) + + this.emit('error', { err }) + } + + private onFFmpegEnded () { + if (this.ended) return + + this.ended = true + this.emit('end') + } + + private buildFFmpegLive () { + return new FFmpegLive(getFFmpegCommandWrapperOptions('live', VideoTranscodingProfilesManager.Instance.getAvailableEncoders())) + } +} diff --git a/server/lib/live/shared/transcoding-wrapper/index.ts b/server/lib/live/shared/transcoding-wrapper/index.ts new file mode 100644 index 000000000..ae28fa1ca --- /dev/null +++ b/server/lib/live/shared/transcoding-wrapper/index.ts @@ -0,0 +1,3 @@ +export * from './abstract-transcoding-wrapper' +export * from './ffmpeg-transcoding-wrapper' +export * from './remote-transcoding-wrapper' diff --git a/server/lib/live/shared/transcoding-wrapper/remote-transcoding-wrapper.ts b/server/lib/live/shared/transcoding-wrapper/remote-transcoding-wrapper.ts new file mode 100644 index 000000000..345eaf442 --- /dev/null +++ b/server/lib/live/shared/transcoding-wrapper/remote-transcoding-wrapper.ts @@ -0,0 +1,20 @@ +import { LiveRTMPHLSTranscodingJobHandler } from '@server/lib/runners' +import { AbstractTranscodingWrapper } from './abstract-transcoding-wrapper' + +export class RemoteTranscodingWrapper extends AbstractTranscodingWrapper { + async run () { + await new LiveRTMPHLSTranscodingJobHandler().create({ + rtmpUrl: this.inputUrl, + toTranscode: this.toTranscode, + video: this.videoLive.Video, + outputDirectory: this.outDirectory, + playlist: this.streamingPlaylist, + segmentListSize: this.segmentListSize, + segmentDuration: this.segmentDuration + }) + } + + abort () { + this.emit('end') + } +} -- cgit v1.2.3