]>
Commit | Line | Data |
---|---|---|
bcd1c9e1 | 1 | import { AsyncQueue, forever, queue } from 'async' |
69818c93 | 2 | import * as Sequelize from 'sequelize' |
571389d4 | 3 | import { JobCategory } from '../../../shared' |
65fcc311 | 4 | import { logger } from '../../helpers' |
571389d4 | 5 | import { database as db, JOB_STATES, JOBS_FETCH_LIMIT_PER_CYCLE, JOBS_FETCHING_INTERVAL } from '../../initializers' |
69818c93 | 6 | import { JobInstance } from '../../models' |
69818c93 | 7 | |
571389d4 C |
8 | export interface JobHandler<P, T> { |
9 | process (data: object, jobId: number): Promise<T> | |
e4f97bab | 10 | onError (err: Error, jobId: number) |
8e10cf1a | 11 | onSuccess (jobId: number, jobResult: T, jobScheduler: JobScheduler<P, T>): Promise<any> |
e4f97bab | 12 | } |
69818c93 | 13 | type JobQueueCallback = (err: Error) => void |
65fcc311 | 14 | |
571389d4 | 15 | class JobScheduler<P, T> { |
65fcc311 | 16 | |
e4f97bab C |
17 | constructor ( |
18 | private jobCategory: JobCategory, | |
571389d4 | 19 | private jobHandlers: { [ id: string ]: JobHandler<P, T> } |
e4f97bab | 20 | ) {} |
65fcc311 | 21 | |
f5028693 | 22 | async activate () { |
e4f97bab | 23 | const limit = JOBS_FETCH_LIMIT_PER_CYCLE[this.jobCategory] |
65fcc311 | 24 | |
e4f97bab | 25 | logger.info('Jobs scheduler %s activated.', this.jobCategory) |
65fcc311 | 26 | |
69818c93 | 27 | const jobsQueue = queue<JobInstance, JobQueueCallback>(this.processJob.bind(this)) |
65fcc311 C |
28 | |
29 | // Finish processing jobs from a previous start | |
30 | const state = JOB_STATES.PROCESSING | |
f5028693 | 31 | try { |
e4f97bab | 32 | const jobs = await db.Job.listWithLimitByCategory(limit, state, this.jobCategory) |
f5028693 C |
33 | |
34 | this.enqueueJobs(jobsQueue, jobs) | |
35 | } catch (err) { | |
36 | logger.error('Cannot list pending jobs.', err) | |
37 | } | |
38 | ||
39 | forever( | |
40 | async next => { | |
41 | if (jobsQueue.length() !== 0) { | |
42 | // Finish processing the queue first | |
43 | return setTimeout(next, JOBS_FETCHING_INTERVAL) | |
44 | } | |
45 | ||
46 | const state = JOB_STATES.PENDING | |
47 | try { | |
e4f97bab | 48 | const jobs = await db.Job.listWithLimitByCategory(limit, state, this.jobCategory) |
f5028693 C |
49 | |
50 | this.enqueueJobs(jobsQueue, jobs) | |
51 | } catch (err) { | |
52 | logger.error('Cannot list pending jobs.', err) | |
53 | } | |
54 | ||
55 | // Optimization: we could use "drain" from queue object | |
56 | return setTimeout(next, JOBS_FETCHING_INTERVAL) | |
57 | }, | |
58 | ||
59 | err => logger.error('Error in job scheduler queue.', err) | |
60 | ) | |
65fcc311 C |
61 | } |
62 | ||
571389d4 | 63 | createJob (transaction: Sequelize.Transaction, handlerName: string, handlerInputData: P) { |
65fcc311 C |
64 | const createQuery = { |
65 | state: JOB_STATES.PENDING, | |
571389d4 | 66 | category: this.jobCategory, |
65fcc311 C |
67 | handlerName, |
68 | handlerInputData | |
69 | } | |
571389d4 | 70 | |
65fcc311 C |
71 | const options = { transaction } |
72 | ||
6fcd19ba | 73 | return db.Job.create(createQuery, options) |
65fcc311 C |
74 | } |
75 | ||
6fcd19ba C |
76 | private enqueueJobs (jobsQueue: AsyncQueue<JobInstance>, jobs: JobInstance[]) { |
77 | jobs.forEach(job => jobsQueue.push(job)) | |
65fcc311 C |
78 | } |
79 | ||
f5028693 | 80 | private async processJob (job: JobInstance, callback: (err: Error) => void) { |
e4f97bab | 81 | const jobHandler = this.jobHandlers[job.handlerName] |
6fcd19ba | 82 | if (jobHandler === undefined) { |
350e31d6 C |
83 | const errorString = 'Unknown job handler ' + job.handlerName + ' for job ' + job.id |
84 | logger.error(errorString) | |
85 | ||
86 | const error = new Error(errorString) | |
87 | await this.onJobError(jobHandler, job, error) | |
88 | return callback(error) | |
6fcd19ba | 89 | } |
65fcc311 C |
90 | |
91 | logger.info('Processing job %d with handler %s.', job.id, job.handlerName) | |
92 | ||
93 | job.state = JOB_STATES.PROCESSING | |
f5028693 C |
94 | await job.save() |
95 | ||
96 | try { | |
571389d4 | 97 | const result: T = await jobHandler.process(job.handlerInputData, job.id) |
f5028693 C |
98 | await this.onJobSuccess(jobHandler, job, result) |
99 | } catch (err) { | |
100 | logger.error('Error in job handler %s.', job.handlerName, err) | |
101 | ||
102 | try { | |
103 | await this.onJobError(jobHandler, job, err) | |
104 | } catch (innerErr) { | |
105 | this.cannotSaveJobError(innerErr) | |
106 | return callback(innerErr) | |
107 | } | |
108 | } | |
109 | ||
350e31d6 | 110 | return callback(null) |
65fcc311 C |
111 | } |
112 | ||
571389d4 | 113 | private async onJobError (jobHandler: JobHandler<P, T>, job: JobInstance, err: Error) { |
65fcc311 C |
114 | job.state = JOB_STATES.ERROR |
115 | ||
f5028693 C |
116 | try { |
117 | await job.save() | |
350e31d6 | 118 | if (jobHandler) await jobHandler.onError(err, job.id) |
f5028693 C |
119 | } catch (err) { |
120 | this.cannotSaveJobError(err) | |
121 | } | |
65fcc311 C |
122 | } |
123 | ||
571389d4 | 124 | private async onJobSuccess (jobHandler: JobHandler<P, T>, job: JobInstance, jobResult: T) { |
65fcc311 C |
125 | job.state = JOB_STATES.SUCCESS |
126 | ||
f5028693 C |
127 | try { |
128 | await job.save() | |
8e10cf1a | 129 | await jobHandler.onSuccess(job.id, jobResult, this) |
f5028693 C |
130 | } catch (err) { |
131 | this.cannotSaveJobError(err) | |
132 | } | |
65fcc311 C |
133 | } |
134 | ||
6fcd19ba | 135 | private cannotSaveJobError (err: Error) { |
ad0997ad | 136 | logger.error('Cannot save new job state.', err) |
65fcc311 C |
137 | } |
138 | } | |
139 | ||
140 | // --------------------------------------------------------------------------- | |
141 | ||
142 | export { | |
143 | JobScheduler | |
144 | } |