1 import * as kue from 'kue'
2 import { JobType, JobState } from '../../../shared/models'
3 import { logger } from '../../helpers/logger'
4 import { CONFIG, JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY } from '../../initializers'
5 import { ActivitypubHttpBroadcastPayload, processActivityPubHttpBroadcast } from './handlers/activitypub-http-broadcast'
6 import { ActivitypubHttpFetcherPayload, processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher'
7 import { ActivitypubHttpUnicastPayload, processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast'
8 import { processVideoFile, VideoFilePayload } from './handlers/video-file'
10 type CreateJobArgument =
11 { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
12 { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } |
13 { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } |
14 { type: 'video-file', payload: VideoFilePayload }
16 const handlers: { [ id in JobType ]: (job: kue.Job) => Promise<any>} = {
17 'activitypub-http-broadcast': processActivityPubHttpBroadcast,
18 'activitypub-http-unicast': processActivityPubHttpUnicast,
19 'activitypub-http-fetcher': processActivityPubHttpFetcher,
20 'video-file': processVideoFile
25 private static instance: JobQueue
27 private jobQueue: kue.Queue
28 private initialized = false
30 private constructor () {}
33 // Already initialized
34 if (this.initialized === true) return
35 this.initialized = true
37 this.jobQueue = kue.createQueue({
38 prefix: 'q-' + CONFIG.WEBSERVER.HOST,
40 host: CONFIG.REDIS.HOSTNAME,
41 port: CONFIG.REDIS.PORT,
42 auth: CONFIG.REDIS.AUTH
46 this.jobQueue.on('error', err => {
47 logger.error('Error in job queue.', err)
50 this.jobQueue.watchStuckJobs(5000)
52 for (const handlerName of Object.keys(handlers)) {
53 this.jobQueue.process(handlerName, JOB_CONCURRENCY[handlerName], async (job, done) => {
55 const res = await handlers[ handlerName ](job)
56 return done(null, res)
64 createJob (obj: CreateJobArgument, priority = 'normal') {
65 return new Promise((res, rej) => {
67 .create(obj.type, obj.payload)
69 .attempts(JOB_ATTEMPTS[obj.type])
70 .backoff({ type: 'exponential' })
72 if (err) return rej(err)
79 listForApi (state: JobState, start: number, count: number, sort: string) {
80 return new Promise<kue.Job[]>((res, rej) => {
81 kue.Job.rangeByState(state, start, count, sort, (err, jobs) => {
82 if (err) return rej(err)
89 count (state: JobState) {
90 return new Promise<number>((res, rej) => {
91 this.jobQueue[state + 'Count']((err, total) => {
92 if (err) return rej(err)
100 const now = new Date().getTime()
101 kue.Job.rangeByState('complete', 0, -1, 'asc', (err, jobs) => {
103 logger.error('Cannot get jobs when removing old jobs.', err)
107 for (const job of jobs) {
108 if (now - job.created_at > JOB_COMPLETED_LIFETIME) {
115 static get Instance () {
116 return this.instance || (this.instance = new this())
120 // ---------------------------------------------------------------------------