aboutsummaryrefslogtreecommitdiffhomepage
path: root/server/lib/job-queue/job-queue.ts
diff options
context:
space:
mode:
Diffstat (limited to 'server/lib/job-queue/job-queue.ts')
-rw-r--r--server/lib/job-queue/job-queue.ts19
1 files changed, 15 insertions, 4 deletions
diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts
index 281e2e51a..3970d48b7 100644
--- a/server/lib/job-queue/job-queue.ts
+++ b/server/lib/job-queue/job-queue.ts
@@ -22,6 +22,7 @@ import {
22 ActivitypubHttpFetcherPayload, 22 ActivitypubHttpFetcherPayload,
23 ActivitypubHttpUnicastPayload, 23 ActivitypubHttpUnicastPayload,
24 ActorKeysPayload, 24 ActorKeysPayload,
25 AfterVideoChannelImportPayload,
25 DeleteResumableUploadMetaFilePayload, 26 DeleteResumableUploadMetaFilePayload,
26 EmailPayload, 27 EmailPayload,
27 FederateVideoPayload, 28 FederateVideoPayload,
@@ -31,6 +32,7 @@ import {
31 MoveObjectStoragePayload, 32 MoveObjectStoragePayload,
32 NotifyPayload, 33 NotifyPayload,
33 RefreshPayload, 34 RefreshPayload,
35 VideoChannelImportPayload,
34 VideoFileImportPayload, 36 VideoFileImportPayload,
35 VideoImportPayload, 37 VideoImportPayload,
36 VideoLiveEndingPayload, 38 VideoLiveEndingPayload,
@@ -53,12 +55,14 @@ import { processFederateVideo } from './handlers/federate-video'
53import { processManageVideoTorrent } from './handlers/manage-video-torrent' 55import { processManageVideoTorrent } from './handlers/manage-video-torrent'
54import { onMoveToObjectStorageFailure, processMoveToObjectStorage } from './handlers/move-to-object-storage' 56import { onMoveToObjectStorageFailure, processMoveToObjectStorage } from './handlers/move-to-object-storage'
55import { processNotify } from './handlers/notify' 57import { processNotify } from './handlers/notify'
58import { processVideoChannelImport } from './handlers/video-channel-import'
56import { processVideoFileImport } from './handlers/video-file-import' 59import { processVideoFileImport } from './handlers/video-file-import'
57import { processVideoImport } from './handlers/video-import' 60import { processVideoImport } from './handlers/video-import'
58import { processVideoLiveEnding } from './handlers/video-live-ending' 61import { processVideoLiveEnding } from './handlers/video-live-ending'
59import { processVideoStudioEdition } from './handlers/video-studio-edition' 62import { processVideoStudioEdition } from './handlers/video-studio-edition'
60import { processVideoTranscoding } from './handlers/video-transcoding' 63import { processVideoTranscoding } from './handlers/video-transcoding'
61import { processVideosViewsStats } from './handlers/video-views-stats' 64import { processVideosViewsStats } from './handlers/video-views-stats'
65import { processAfterVideoChannelImport } from './handlers/after-video-channel-import'
62 66
63export type CreateJobArgument = 67export type CreateJobArgument =
64 { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } | 68 { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
@@ -79,6 +83,9 @@ export type CreateJobArgument =
79 { type: 'delete-resumable-upload-meta-file', payload: DeleteResumableUploadMetaFilePayload } | 83 { type: 'delete-resumable-upload-meta-file', payload: DeleteResumableUploadMetaFilePayload } |
80 { type: 'video-studio-edition', payload: VideoStudioEditionPayload } | 84 { type: 'video-studio-edition', payload: VideoStudioEditionPayload } |
81 { type: 'manage-video-torrent', payload: ManageVideoTorrentPayload } | 85 { type: 'manage-video-torrent', payload: ManageVideoTorrentPayload } |
86 { type: 'move-to-object-storage', payload: MoveObjectStoragePayload } |
87 { type: 'video-channel-import', payload: VideoChannelImportPayload } |
88 { type: 'after-video-channel-import', payload: AfterVideoChannelImportPayload } |
82 { type: 'notify', payload: NotifyPayload } | 89 { type: 'notify', payload: NotifyPayload } |
83 { type: 'move-to-object-storage', payload: MoveObjectStoragePayload } | 90 { type: 'move-to-object-storage', payload: MoveObjectStoragePayload } |
84 { type: 'federate-video', payload: FederateVideoPayload } 91 { type: 'federate-video', payload: FederateVideoPayload }
@@ -106,8 +113,10 @@ const handlers: { [id in JobType]: (job: Job) => Promise<any> } = {
106 'video-redundancy': processVideoRedundancy, 113 'video-redundancy': processVideoRedundancy,
107 'move-to-object-storage': processMoveToObjectStorage, 114 'move-to-object-storage': processMoveToObjectStorage,
108 'manage-video-torrent': processManageVideoTorrent, 115 'manage-video-torrent': processManageVideoTorrent,
109 'notify': processNotify,
110 'video-studio-edition': processVideoStudioEdition, 116 'video-studio-edition': processVideoStudioEdition,
117 'video-channel-import': processVideoChannelImport,
118 'after-video-channel-import': processAfterVideoChannelImport,
119 'notify': processNotify,
111 'federate-video': processFederateVideo 120 'federate-video': processFederateVideo
112} 121}
113 122
@@ -134,6 +143,8 @@ const jobTypes: JobType[] = [
134 'move-to-object-storage', 143 'move-to-object-storage',
135 'manage-video-torrent', 144 'manage-video-torrent',
136 'video-studio-edition', 145 'video-studio-edition',
146 'video-channel-import',
147 'after-video-channel-import',
137 'notify', 148 'notify',
138 'federate-video' 149 'federate-video'
139] 150]
@@ -306,7 +317,7 @@ class JobQueue {
306 .catch(err => logger.error('Cannot create job.', { err, options })) 317 .catch(err => logger.error('Cannot create job.', { err, options }))
307 } 318 }
308 319
309 async createJob (options: CreateJobArgument & CreateJobOptions) { 320 createJob (options: CreateJobArgument & CreateJobOptions) {
310 const queue: Queue = this.queues[options.type] 321 const queue: Queue = this.queues[options.type]
311 if (queue === undefined) { 322 if (queue === undefined) {
312 logger.error('Unknown queue %s: cannot create job.', options.type) 323 logger.error('Unknown queue %s: cannot create job.', options.type)
@@ -318,7 +329,7 @@ class JobQueue {
318 return queue.add('job', options.payload, jobOptions) 329 return queue.add('job', options.payload, jobOptions)
319 } 330 }
320 331
321 async createSequentialJobFlow (...jobs: ((CreateJobArgument & CreateJobOptions) | undefined)[]) { 332 createSequentialJobFlow (...jobs: ((CreateJobArgument & CreateJobOptions) | undefined)[]) {
322 let lastJob: FlowJob 333 let lastJob: FlowJob
323 334
324 for (const job of jobs) { 335 for (const job of jobs) {
@@ -336,7 +347,7 @@ class JobQueue {
336 return this.flowProducer.add(lastJob) 347 return this.flowProducer.add(lastJob)
337 } 348 }
338 349
339 async createJobWithChildren (parent: CreateJobArgument & CreateJobOptions, children: (CreateJobArgument & CreateJobOptions)[]) { 350 createJobWithChildren (parent: CreateJobArgument & CreateJobOptions, children: (CreateJobArgument & CreateJobOptions)[]) {
340 return this.flowProducer.add({ 351 return this.flowProducer.add({
341 ...this.buildJobFlowOption(parent), 352 ...this.buildJobFlowOption(parent),
342 353