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