diff options
author | Chocobozzz <me@florianbigard.com> | 2018-01-25 15:05:18 +0100 |
---|---|---|
committer | Chocobozzz <me@florianbigard.com> | 2018-01-25 18:41:17 +0100 |
commit | 94a5ff8a4a75d75bb9df542a39ce8769e7a7e6a4 (patch) | |
tree | 32a9148e0e4567f0c4ffae0412cbed20b84e8873 /server/lib/job-queue/handlers/video-file.ts | |
parent | d765fafc3faf0db9818eb1a07161df1cb1bc0efa (diff) | |
download | PeerTube-94a5ff8a4a75d75bb9df542a39ce8769e7a7e6a4.tar.gz PeerTube-94a5ff8a4a75d75bb9df542a39ce8769e7a7e6a4.tar.zst PeerTube-94a5ff8a4a75d75bb9df542a39ce8769e7a7e6a4.zip |
Move job queue to redis
We'll use it as cache in the future.
/!\ You'll loose your old jobs (pending jobs too) so upgrade only when
you don't have pending job anymore.
Diffstat (limited to 'server/lib/job-queue/handlers/video-file.ts')
-rw-r--r-- | server/lib/job-queue/handlers/video-file.ts | 110 |
1 files changed, 110 insertions, 0 deletions
diff --git a/server/lib/job-queue/handlers/video-file.ts b/server/lib/job-queue/handlers/video-file.ts new file mode 100644 index 000000000..5294483bd --- /dev/null +++ b/server/lib/job-queue/handlers/video-file.ts | |||
@@ -0,0 +1,110 @@ | |||
1 | import * as kue from 'kue' | ||
2 | import { VideoResolution } from '../../../../shared' | ||
3 | import { VideoPrivacy } from '../../../../shared/models/videos' | ||
4 | import { logger } from '../../../helpers/logger' | ||
5 | import { computeResolutionsToTranscode } from '../../../helpers/utils' | ||
6 | import { sequelizeTypescript } from '../../../initializers' | ||
7 | import { VideoModel } from '../../../models/video/video' | ||
8 | import { shareVideoByServerAndChannel } from '../../activitypub' | ||
9 | import { sendCreateVideo, sendUpdateVideo } from '../../activitypub/send' | ||
10 | import { JobQueue } from '../job-queue' | ||
11 | |||
12 | export type VideoFilePayload = { | ||
13 | videoUUID: string | ||
14 | resolution?: VideoResolution | ||
15 | } | ||
16 | |||
17 | async function processVideoFile (job: kue.Job) { | ||
18 | const payload = job.data as VideoFilePayload | ||
19 | logger.info('Processing video file in job %d.', job.id) | ||
20 | |||
21 | const video = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(payload.videoUUID) | ||
22 | // No video, maybe deleted? | ||
23 | if (!video) { | ||
24 | logger.info('Do not process job %d, video does not exist.', job.id, { videoUUID: video.uuid }) | ||
25 | return undefined | ||
26 | } | ||
27 | |||
28 | // Transcoding in other resolution | ||
29 | if (payload.resolution) { | ||
30 | await video.transcodeOriginalVideofile(payload.resolution) | ||
31 | await onVideoFileTranscoderSuccess(video) | ||
32 | } else { | ||
33 | await video.optimizeOriginalVideofile() | ||
34 | await onVideoFileOptimizerSuccess(video) | ||
35 | } | ||
36 | |||
37 | return video | ||
38 | } | ||
39 | |||
40 | async function onVideoFileTranscoderSuccess (video: VideoModel) { | ||
41 | if (video === undefined) return undefined | ||
42 | |||
43 | // Maybe the video changed in database, refresh it | ||
44 | const videoDatabase = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(video.uuid) | ||
45 | // Video does not exist anymore | ||
46 | if (!videoDatabase) return undefined | ||
47 | |||
48 | if (video.privacy !== VideoPrivacy.PRIVATE) { | ||
49 | await sendUpdateVideo(video, undefined) | ||
50 | } | ||
51 | |||
52 | return undefined | ||
53 | } | ||
54 | |||
55 | async function onVideoFileOptimizerSuccess (video: VideoModel) { | ||
56 | if (video === undefined) return undefined | ||
57 | |||
58 | // Maybe the video changed in database, refresh it | ||
59 | const videoDatabase = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(video.uuid) | ||
60 | // Video does not exist anymore | ||
61 | if (!videoDatabase) return undefined | ||
62 | |||
63 | if (video.privacy !== VideoPrivacy.PRIVATE) { | ||
64 | // Now we'll add the video's meta data to our followers | ||
65 | await sendCreateVideo(video, undefined) | ||
66 | await shareVideoByServerAndChannel(video, undefined) | ||
67 | } | ||
68 | |||
69 | const originalFileHeight = await videoDatabase.getOriginalFileHeight() | ||
70 | |||
71 | // Create transcoding jobs if there are enabled resolutions | ||
72 | const resolutionsEnabled = computeResolutionsToTranscode(originalFileHeight) | ||
73 | logger.info( | ||
74 | 'Resolutions computed for video %s and origin file height of %d.', videoDatabase.uuid, originalFileHeight, | ||
75 | { resolutions: resolutionsEnabled } | ||
76 | ) | ||
77 | |||
78 | if (resolutionsEnabled.length !== 0) { | ||
79 | try { | ||
80 | await sequelizeTypescript.transaction(async t => { | ||
81 | const tasks: Promise<any>[] = [] | ||
82 | |||
83 | for (const resolution of resolutionsEnabled) { | ||
84 | const dataInput = { | ||
85 | videoUUID: videoDatabase.uuid, | ||
86 | resolution | ||
87 | } | ||
88 | |||
89 | const p = JobQueue.Instance.createJob({ type: 'video-file', payload: dataInput }) | ||
90 | tasks.push(p) | ||
91 | } | ||
92 | |||
93 | await Promise.all(tasks) | ||
94 | }) | ||
95 | |||
96 | logger.info('Transcoding jobs created for uuid %s.', videoDatabase.uuid, { resolutionsEnabled }) | ||
97 | } catch (err) { | ||
98 | logger.warn('Cannot transcode the video.', err) | ||
99 | } | ||
100 | } else { | ||
101 | logger.info('No transcoding jobs created for video %s (no resolutions enabled).') | ||
102 | return undefined | ||
103 | } | ||
104 | } | ||
105 | |||
106 | // --------------------------------------------------------------------------- | ||
107 | |||
108 | export { | ||
109 | processVideoFile | ||
110 | } | ||