X-Git-Url: https://git.immae.eu/?a=blobdiff_plain;f=server%2Flib%2Fjob-queue%2Fjob-queue.ts;h=4cda12b5754c5ca411384656eaaf2f4edafc192b;hb=e1a570abff3ebf375433e58e7362d56bd32d4cd8;hp=42e8347b1f663650c5aed7042c5e02cbd1b9ffcc;hpb=74d249bc1346c7cfaac7ee49bebbebcf2a01f82a;p=github%2FChocobozzz%2FPeerTube.git diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts index 42e8347b1..4cda12b57 100644 --- a/server/lib/job-queue/job-queue.ts +++ b/server/lib/job-queue/job-queue.ts @@ -1,4 +1,4 @@ -import * as Bull from 'bull' +import Bull, { Job, JobOptions, Queue } from 'bull' import { jobStates } from '@server/helpers/custom-validators/jobs' import { CONFIG } from '@server/initializers/config' import { processVideoRedundancy } from '@server/lib/job-queue/handlers/video-redundancy' @@ -11,6 +11,7 @@ import { EmailPayload, JobState, JobType, + MoveObjectStoragePayload, RefreshPayload, VideoFileImportPayload, VideoImportPayload, @@ -29,6 +30,7 @@ import { processActivityPubHttpUnicast } from './handlers/activitypub-http-unica import { refreshAPObject } from './handlers/activitypub-refresher' import { processActorKeys } from './handlers/actor-keys' import { processEmail } from './handlers/email' +import { processMoveToObjectStorage } from './handlers/move-to-object-storage' import { processVideoFileImport } from './handlers/video-file-import' import { processVideoImport } from './handlers/video-import' import { processVideoLiveEnding } from './handlers/video-live-ending' @@ -49,14 +51,15 @@ type CreateJobArgument = { type: 'videos-views', payload: {} } | { type: 'video-live-ending', payload: VideoLiveEndingPayload } | { type: 'actor-keys', payload: ActorKeysPayload } | - { type: 'video-redundancy', payload: VideoRedundancyPayload } + { type: 'video-redundancy', payload: VideoRedundancyPayload } | + { type: 'move-to-object-storage', payload: MoveObjectStoragePayload } -type CreateJobOptions = { +export type CreateJobOptions = { delay?: number priority?: number } -const handlers: { [id in JobType]: (job: Bull.Job) => Promise } = { +const handlers: { [id in JobType]: (job: Job) => Promise } = { 'activitypub-http-broadcast': processActivityPubHttpBroadcast, 'activitypub-http-unicast': processActivityPubHttpUnicast, 'activitypub-http-fetcher': processActivityPubHttpFetcher, @@ -70,7 +73,8 @@ const handlers: { [id in JobType]: (job: Bull.Job) => Promise } = { 'activitypub-refresher': refreshAPObject, 'video-live-ending': processVideoLiveEnding, 'actor-keys': processActorKeys, - 'video-redundancy': processVideoRedundancy + 'video-redundancy': processVideoRedundancy, + 'move-to-object-storage': processMoveToObjectStorage } const jobTypes: JobType[] = [ @@ -87,14 +91,15 @@ const jobTypes: JobType[] = [ 'activitypub-refresher', 'video-redundancy', 'actor-keys', - 'video-live-ending' + 'video-live-ending', + 'move-to-object-storage' ] class JobQueue { private static instance: JobQueue - private queues: { [id in JobType]?: Bull.Queue } = {} + private queues: { [id in JobType]?: Queue } = {} private initialized = false private jobRedisPrefix: string @@ -155,7 +160,7 @@ class JobQueue { return } - const jobArgs: Bull.JobOptions = { + const jobArgs: JobOptions = { backoff: { delay: 60 * 1000, type: 'exponential' }, attempts: JOB_ATTEMPTS[obj.type], timeout: JOB_TTL[obj.type], @@ -172,11 +177,11 @@ class JobQueue { count: number asc?: boolean jobType: JobType - }): Promise { + }): Promise { const { state, start, count, asc, jobType } = options const states = state ? [ state ] : jobStates - let results: Bull.Job[] = [] + let results: Job[] = [] const filteredJobTypes = this.filterJobTypes(jobType)