diff options
Diffstat (limited to 'server/lib/job-queue/job-queue.ts')
-rw-r--r-- | server/lib/job-queue/job-queue.ts | 19 |
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' | |||
53 | import { processManageVideoTorrent } from './handlers/manage-video-torrent' | 55 | import { processManageVideoTorrent } from './handlers/manage-video-torrent' |
54 | import { onMoveToObjectStorageFailure, processMoveToObjectStorage } from './handlers/move-to-object-storage' | 56 | import { onMoveToObjectStorageFailure, processMoveToObjectStorage } from './handlers/move-to-object-storage' |
55 | import { processNotify } from './handlers/notify' | 57 | import { processNotify } from './handlers/notify' |
58 | import { processVideoChannelImport } from './handlers/video-channel-import' | ||
56 | import { processVideoFileImport } from './handlers/video-file-import' | 59 | import { processVideoFileImport } from './handlers/video-file-import' |
57 | import { processVideoImport } from './handlers/video-import' | 60 | import { processVideoImport } from './handlers/video-import' |
58 | import { processVideoLiveEnding } from './handlers/video-live-ending' | 61 | import { processVideoLiveEnding } from './handlers/video-live-ending' |
59 | import { processVideoStudioEdition } from './handlers/video-studio-edition' | 62 | import { processVideoStudioEdition } from './handlers/video-studio-edition' |
60 | import { processVideoTranscoding } from './handlers/video-transcoding' | 63 | import { processVideoTranscoding } from './handlers/video-transcoding' |
61 | import { processVideosViewsStats } from './handlers/video-views-stats' | 64 | import { processVideosViewsStats } from './handlers/video-views-stats' |
65 | import { processAfterVideoChannelImport } from './handlers/after-video-channel-import' | ||
62 | 66 | ||
63 | export type CreateJobArgument = | 67 | export 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 | ||