123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246 |
- import * as Bull from 'bull'
- 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 { JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_TTL, REPEAT_JOBS, WEBSERVER } from '../../initializers/constants'
- 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'
- type CreateJobArgument =
- { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
- { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } |
- { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } |
- { type: 'activitypub-follow', payload: ActivitypubFollowPayload } |
- { type: 'video-file-import', payload: VideoFileImportPayload } |
- { type: 'video-transcoding', payload: VideoTranscodingPayload } |
- { type: 'email', payload: EmailPayload } |
- { type: 'video-import', payload: VideoImportPayload } |
- { type: 'activitypub-refresher', payload: RefreshPayload } |
- { type: 'videos-views', payload: {} } |
- { type: 'video-live-ending', payload: VideoLiveEndingPayload } |
- { type: 'video-redundancy', payload: VideoRedundancyPayload }
- type CreateJobOptions = {
- delay?: number
- }
- const handlers: { [id in JobType]: (job: Bull.Job) => Promise<any> } = {
- 'activitypub-http-broadcast': processActivityPubHttpBroadcast,
- 'activitypub-http-unicast': processActivityPubHttpUnicast,
- 'activitypub-http-fetcher': processActivityPubHttpFetcher,
- 'activitypub-follow': processActivityPubFollow,
- 'video-file-import': processVideoFileImport,
- 'video-transcoding': processVideoTranscoding,
- 'email': processEmail,
- 'video-import': processVideoImport,
- 'videos-views': processVideosViews,
- 'activitypub-refresher': refreshAPObject,
- 'video-live-ending': processVideoLiveEnding,
- 'video-redundancy': processVideoRedundancy
- }
- const jobTypes: JobType[] = [
- 'activitypub-follow',
- 'activitypub-http-broadcast',
- 'activitypub-http-fetcher',
- 'activitypub-http-unicast',
- 'email',
- 'video-transcoding',
- 'video-file-import',
- 'video-import',
- 'videos-views',
- 'activitypub-refresher',
- 'video-redundancy',
- 'video-live-ending'
- ]
- class JobQueue {
- private static instance: JobQueue
- private queues: { [id in JobType]?: Bull.Queue } = {}
- private initialized = false
- private jobRedisPrefix: string
- private constructor () {
- }
- init () {
- // Already initialized
- if (this.initialized === true) return
- this.initialized = true
- this.jobRedisPrefix = 'bull-' + WEBSERVER.HOST
- const queueOptions = {
- prefix: this.jobRedisPrefix,
- redis: Redis.getRedisClientOptions(),
- settings: {
- maxStalledCount: 10 // transcoding could be long, so jobs can often be interrupted by restarts
- }
- }
- for (const handlerName of Object.keys(handlers)) {
- const queue = new Bull(handlerName, queueOptions)
- const handler = handlers[handlerName]
- queue.process(JOB_CONCURRENCY[handlerName], handler)
- .catch(err => logger.error('Error in job queue processor %s.', handlerName, { err }))
- queue.on('failed', (job, err) => {
- logger.error('Cannot execute job %d in queue %s.', job.id, handlerName, { payload: job.data, err })
- })
- queue.on('error', err => {
- logger.error('Error in job queue %s.', handlerName, { err })
- })
- this.queues[handlerName] = queue
- }
- this.addRepeatableJobs()
- }
- terminate () {
- for (const queueName of Object.keys(this.queues)) {
- const queue = this.queues[queueName]
- queue.close()
- }
- }
- 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)
- return
- }
- const jobArgs: Bull.JobOptions = {
- backoff: { delay: 60 * 1000, type: 'exponential' },
- attempts: JOB_ATTEMPTS[obj.type],
- timeout: JOB_TTL[obj.type],
- delay: options.delay
- }
- return queue.add(obj.payload, jobArgs)
- }
- async listForApi (options: {
- state?: JobState
- start: number
- count: number
- asc?: boolean
- jobType: JobType
- }): Promise<Bull.Job[]> {
- const { state, start, count, asc, jobType } = options
- const states = state ? [ state ] : jobStates
- let results: Bull.Job[] = []
- 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
- }
- const jobs = await queue.getJobs(states, 0, start + count, asc)
- results = results.concat(jobs)
- }
- results.sort((j1: any, j2: any) => {
- if (j1.timestamp < j2.timestamp) return -1
- else if (j1.timestamp === j2.timestamp) return 0
- return 1
- })
- if (asc === false) results.reverse()
- return results.slice(start, start + count)
- }
- async count (state: JobState, jobType?: JobType): Promise<number> {
- const states = state ? [ state ] : jobStates
- let total = 0
- 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
- }
- const counts = await queue.getJobCounts()
- for (const s of states) {
- total += counts[s]
- }
- }
- return total
- }
- async removeOldJobs () {
- for (const key of Object.keys(this.queues)) {
- const queue = this.queues[key]
- await queue.clean(JOB_COMPLETED_LIFETIME, 'completed')
- }
- }
- 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 () {
- return this.instance || (this.instance = new this())
- }
- }
- // ---------------------------------------------------------------------------
- export {
- jobTypes,
- JobQueue
- }
|