1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
import { Job } from 'bullmq'
import { onTranscodingEnded } from '@server/lib/transcoding/ended-transcoding'
import { generateHlsPlaylistResolution } from '@server/lib/transcoding/hls-transcoding'
import { mergeAudioVideofile, optimizeOriginalVideofile, transcodeNewWebTorrentResolution } from '@server/lib/transcoding/web-transcoding'
import { removeAllWebTorrentFiles } from '@server/lib/video-file'
import { VideoPathManager } from '@server/lib/video-path-manager'
import { moveToFailedTranscodingState } from '@server/lib/video-state'
import { UserModel } from '@server/models/user/user'
import { VideoJobInfoModel } from '@server/models/video/video-job-info'
import { MUser, MUserId, MVideoFullLight } from '@server/types/models'
import {
HLSTranscodingPayload,
MergeAudioTranscodingPayload,
NewWebTorrentResolutionTranscodingPayload,
OptimizeTranscodingPayload,
VideoTranscodingPayload
} from '@shared/models'
import { logger, loggerTagsFactory } from '../../../helpers/logger'
import { VideoModel } from '../../../models/video/video'
type HandlerFunction = (job: Job, payload: VideoTranscodingPayload, video: MVideoFullLight, user: MUser) => Promise<void>
const handlers: { [ id in VideoTranscodingPayload['type'] ]: HandlerFunction } = {
'new-resolution-to-hls': handleHLSJob,
'new-resolution-to-webtorrent': handleNewWebTorrentResolutionJob,
'merge-audio-to-webtorrent': handleWebTorrentMergeAudioJob,
'optimize-to-webtorrent': handleWebTorrentOptimizeJob
}
const lTags = loggerTagsFactory('transcoding')
async function processVideoTranscoding (job: Job) {
const payload = job.data as VideoTranscodingPayload
logger.info('Processing transcoding job %s.', job.id, lTags(payload.videoUUID))
const video = await VideoModel.loadFull(payload.videoUUID)
// No video, maybe deleted?
if (!video) {
logger.info('Do not process job %d, video does not exist.', job.id, lTags(payload.videoUUID))
return undefined
}
const user = await UserModel.loadByChannelActorId(video.VideoChannel.actorId)
const handler = handlers[payload.type]
if (!handler) {
await moveToFailedTranscodingState(video)
await VideoJobInfoModel.decrease(video.uuid, 'pendingTranscode')
throw new Error('Cannot find transcoding handler for ' + payload.type)
}
try {
await handler(job, payload, video, user)
} catch (error) {
await moveToFailedTranscodingState(video)
await VideoJobInfoModel.decrease(video.uuid, 'pendingTranscode')
throw error
}
return video
}
// ---------------------------------------------------------------------------
export {
processVideoTranscoding
}
// ---------------------------------------------------------------------------
// Job handlers
// ---------------------------------------------------------------------------
async function handleWebTorrentMergeAudioJob (job: Job, payload: MergeAudioTranscodingPayload, video: MVideoFullLight, user: MUserId) {
logger.info('Handling merge audio transcoding job for %s.', video.uuid, lTags(video.uuid))
await mergeAudioVideofile({ video, resolution: payload.resolution, fps: payload.fps, job })
logger.info('Merge audio transcoding job for %s ended.', video.uuid, lTags(video.uuid))
await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: true, video })
}
async function handleWebTorrentOptimizeJob (job: Job, payload: OptimizeTranscodingPayload, video: MVideoFullLight, user: MUserId) {
logger.info('Handling optimize transcoding job for %s.', video.uuid, lTags(video.uuid))
await optimizeOriginalVideofile({ video, inputVideoFile: video.getMaxQualityFile(), quickTranscode: payload.quickTranscode, job })
logger.info('Optimize transcoding job for %s ended.', video.uuid, lTags(video.uuid))
await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: true, video })
}
async function handleNewWebTorrentResolutionJob (job: Job, payload: NewWebTorrentResolutionTranscodingPayload, video: MVideoFullLight) {
logger.info('Handling WebTorrent transcoding job for %s.', video.uuid, lTags(video.uuid))
await transcodeNewWebTorrentResolution({ video, resolution: payload.resolution, fps: payload.fps, job })
logger.info('WebTorrent transcoding job for %s ended.', video.uuid, lTags(video.uuid))
await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: true, video })
}
async function handleHLSJob (job: Job, payload: HLSTranscodingPayload, video: MVideoFullLight) {
logger.info('Handling HLS transcoding job for %s.', video.uuid, lTags(video.uuid))
const videoFileInput = payload.copyCodecs
? video.getWebTorrentFile(payload.resolution)
: video.getMaxQualityFile()
const videoOrStreamingPlaylist = videoFileInput.getVideoOrStreamingPlaylist()
const inputFileMutexReleaser = await VideoPathManager.Instance.lockFiles(video.uuid)
try {
await videoFileInput.getVideo().reload()
await VideoPathManager.Instance.makeAvailableVideoFile(videoFileInput.withVideoOrPlaylist(videoOrStreamingPlaylist), videoInputPath => {
return generateHlsPlaylistResolution({
video,
videoInputPath,
inputFileMutexReleaser,
resolution: payload.resolution,
fps: payload.fps,
copyCodecs: payload.copyCodecs,
job
})
})
} finally {
inputFileMutexReleaser()
}
logger.info('HLS transcoding job for %s ended.', video.uuid, lTags(video.uuid))
if (payload.deleteWebTorrentFiles === true) {
logger.info('Removing WebTorrent files of %s now we have a HLS version of it.', video.uuid, lTags(video.uuid))
await removeAllWebTorrentFiles(video)
}
await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: true, video })
}
|
