X-Git-Url: https://git.immae.eu/?a=blobdiff_plain;f=server%2Flib%2Fjob-queue%2Fjob-queue.ts;h=38b1d6f1ff6ed9d0a5c5056195660f1e18a75314;hb=77d7e851dccf17dcc89e8fcc2db3f655d1e63f95;hp=3c810da98e6d9b3fb4a889cd935aec8d16cabcbd;hpb=97567dd81f508dd6295ac4d73d849aa2ce0a6549;p=github%2FChocobozzz%2FPeerTube.git diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts index 3c810da98..38b1d6f1f 100644 --- a/server/lib/job-queue/job-queue.ts +++ b/server/lib/job-queue/job-queue.ts @@ -1,18 +1,35 @@ import * as Bull from 'bull' -import { JobState, JobType } from '../../../shared/models' +import { jobStates } from '@server/helpers/custom-validators/jobs' +import { processVideoRedundancy } from '@server/lib/job-queue/handlers/video-redundancy' +import { + ActivitypubFollowPayload, + ActivitypubHttpBroadcastPayload, + ActivitypubHttpFetcherPayload, + ActivitypubHttpUnicastPayload, + EmailPayload, + JobState, + JobType, + RefreshPayload, + VideoFileImportPayload, + VideoImportPayload, + VideoLiveEndingPayload, + VideoRedundancyPayload, + VideoTranscodingPayload +} from '../../../shared/models' import { logger } from '../../helpers/logger' -import { Redis } from '../redis' import { JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_TTL, REPEAT_JOBS, WEBSERVER } from '../../initializers/constants' -import { ActivitypubHttpBroadcastPayload, processActivityPubHttpBroadcast } from './handlers/activitypub-http-broadcast' -import { ActivitypubHttpFetcherPayload, processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher' -import { ActivitypubHttpUnicastPayload, processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast' -import { EmailPayload, processEmail } from './handlers/email' -import { processVideoTranscoding, VideoTranscodingPayload } from './handlers/video-transcoding' -import { ActivitypubFollowPayload, processActivityPubFollow } from './handlers/activitypub-follow' -import { processVideoImport, VideoImportPayload } from './handlers/video-import' +import { Redis } from '../redis' +import { processActivityPubFollow } from './handlers/activitypub-follow' +import { processActivityPubHttpBroadcast } from './handlers/activitypub-http-broadcast' +import { processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher' +import { processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast' +import { refreshAPObject } from './handlers/activitypub-refresher' +import { processEmail } from './handlers/email' +import { processVideoFileImport } from './handlers/video-file-import' +import { processVideoImport } from './handlers/video-import' +import { processVideoLiveEnding } from './handlers/video-live-ending' +import { processVideoTranscoding } from './handlers/video-transcoding' import { processVideosViews } from './handlers/video-views' -import { refreshAPObject, RefreshPayload } from './handlers/activitypub-refresher' -import { processVideoFileImport, VideoFileImportPayload } from './handlers/video-file-import' type CreateJobArgument = { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } | @@ -24,20 +41,28 @@ type CreateJobArgument = { type: 'email', payload: EmailPayload } | { type: 'video-import', payload: VideoImportPayload } | { type: 'activitypub-refresher', payload: RefreshPayload } | - { type: 'videos-views', payload: {} } + { type: 'videos-views', payload: {} } | + { type: 'video-live-ending', payload: VideoLiveEndingPayload } | + { type: 'video-redundancy', payload: VideoRedundancyPayload } + +type CreateJobOptions = { + delay?: number + priority?: number +} -const handlers: { [ id in (JobType | 'video-file') ]: (job: Bull.Job) => Promise} = { +const handlers: { [id in JobType]: (job: Bull.Job) => Promise } = { 'activitypub-http-broadcast': processActivityPubHttpBroadcast, 'activitypub-http-unicast': processActivityPubHttpUnicast, 'activitypub-http-fetcher': processActivityPubHttpFetcher, 'activitypub-follow': processActivityPubFollow, 'video-file-import': processVideoFileImport, 'video-transcoding': processVideoTranscoding, - 'video-file': processVideoTranscoding, // TODO: remove it (changed in 1.3) 'email': processEmail, 'video-import': processVideoImport, 'videos-views': processVideosViews, - 'activitypub-refresher': refreshAPObject + 'activitypub-refresher': refreshAPObject, + 'video-live-ending': processVideoLiveEnding, + 'video-redundancy': processVideoRedundancy } const jobTypes: JobType[] = [ @@ -50,20 +75,23 @@ const jobTypes: JobType[] = [ 'video-file-import', 'video-import', 'videos-views', - 'activitypub-refresher' + 'activitypub-refresher', + 'video-redundancy', + 'video-live-ending' ] class JobQueue { private static instance: JobQueue - private queues: { [ id in JobType ]?: Bull.Queue } = {} + private queues: { [id in JobType]?: Bull.Queue } = {} private initialized = false private jobRedisPrefix: string - private constructor () {} + private constructor () { + } - async init () { + init () { // Already initialized if (this.initialized === true) return this.initialized = true @@ -105,35 +133,51 @@ class JobQueue { } } - createJob (obj: CreateJobArgument) { + createJob (obj: CreateJobArgument, options: CreateJobOptions = {}): void { + this.createJobWithPromise(obj, options) + .catch(err => logger.error('Cannot create job.', { err, obj })) + } + + createJobWithPromise (obj: CreateJobArgument, options: CreateJobOptions = {}) { const queue = this.queues[obj.type] if (queue === undefined) { logger.error('Unknown queue %s: cannot create job.', obj.type) - throw Error('Unknown queue, cannot create job') + return } const jobArgs: Bull.JobOptions = { backoff: { delay: 60 * 1000, type: 'exponential' }, attempts: JOB_ATTEMPTS[obj.type], - timeout: JOB_TTL[obj.type] + timeout: JOB_TTL[obj.type], + priority: options.priority, + delay: options.delay } return queue.add(obj.payload, jobArgs) } - async listForApi (state: JobState, start: number, count: number, asc?: boolean): Promise { + async listForApi (options: { + state?: JobState + start: number + count: number + asc?: boolean + jobType: JobType + }): Promise { + const { state, start, count, asc, jobType } = options + + const states = state ? [ state ] : jobStates let results: Bull.Job[] = [] - // TODO: optimize - for (const jobType of jobTypes) { - const queue = this.queues[ jobType ] + const filteredJobTypes = this.filterJobTypes(jobType) + + for (const jobType of filteredJobTypes) { + const queue = this.queues[jobType] if (queue === undefined) { logger.error('Unknown queue %s to list jobs.', jobType) continue } - // FIXME: Bull queue typings does not have getJobs method - const jobs = await (queue as any).getJobs(state, 0, start + count, asc) + const jobs = await queue.getJobs(states, 0, start + count, asc) results = results.concat(jobs) } @@ -149,11 +193,14 @@ class JobQueue { return results.slice(start, start + count) } - async count (state: JobState): Promise { + async count (state: JobState, jobType?: JobType): Promise { + const states = state ? [ state ] : jobStates let total = 0 - for (const type of jobTypes) { - const queue = this.queues[ type ] + const filteredJobTypes = this.filterJobTypes(jobType) + + for (const type of filteredJobTypes) { + const queue = this.queues[type] if (queue === undefined) { logger.error('Unknown queue %s to count jobs.', type) continue @@ -161,7 +208,9 @@ class JobQueue { const counts = await queue.getJobCounts() - total += counts[ state ] + for (const s of states) { + total += counts[s] + } } return total @@ -177,7 +226,13 @@ class JobQueue { private addRepeatableJobs () { this.queues['videos-views'].add({}, { repeat: REPEAT_JOBS['videos-views'] - }) + }).catch(err => logger.error('Cannot add repeatable job.', { err })) + } + + private filterJobTypes (jobType?: JobType) { + if (!jobType) return jobTypes + + return jobTypes.filter(t => t === jobType) } static get Instance () { @@ -188,5 +243,6 @@ class JobQueue { // --------------------------------------------------------------------------- export { + jobTypes, JobQueue }