1 import { forever, queue } from 'async'
3 import { database as db } from '../../initializers/database'
5 JOBS_FETCHING_INTERVAL,
6 JOBS_FETCH_LIMIT_PER_CYCLE,
8 } from '../../initializers'
9 import { logger } from '../../helpers'
10 import { jobHandlers } from './handlers'
14 private static instance: JobScheduler
16 private constructor () { }
18 static get Instance () {
19 return this.instance || (this.instance = new this())
23 const limit = JOBS_FETCH_LIMIT_PER_CYCLE
25 logger.info('Jobs scheduler activated.')
27 const jobsQueue = queue(this.processJob.bind(this))
29 // Finish processing jobs from a previous start
30 const state = JOB_STATES.PROCESSING
31 db.Job.listWithLimit(limit, state, (err, jobs) => {
32 this.enqueueJobs(err, jobsQueue, jobs)
36 if (jobsQueue.length() !== 0) {
37 // Finish processing the queue first
38 return setTimeout(next, JOBS_FETCHING_INTERVAL)
41 const state = JOB_STATES.PENDING
42 db.Job.listWithLimit(limit, state, (err, jobs) => {
44 logger.error('Cannot list pending jobs.', { error: err })
51 // Optimization: we could use "drain" from queue object
52 return setTimeout(next, JOBS_FETCHING_INTERVAL)
56 err => { logger.error('Error in job scheduler queue.', { error: err }) }
61 createJob (transaction, handlerName: string, handlerInputData: object, callback) {
63 state: JOB_STATES.PENDING,
67 const options = { transaction }
69 db.Job.create(createQuery, options).asCallback(callback)
72 private enqueueJobs (err, jobsQueue, jobs) {
74 logger.error('Cannot list pending jobs.', { error: err })
82 private processJob (job, callback) {
83 const jobHandler = jobHandlers[job.handlerName]
85 logger.info('Processing job %d with handler %s.', job.id, job.handlerName)
87 job.state = JOB_STATES.PROCESSING
88 job.save().asCallback(err => {
89 if (err) return this.cannotSaveJobError(err, callback)
91 if (jobHandler === undefined) {
92 logger.error('Unknown job handler for job %s.', jobHandler.handlerName)
96 return jobHandler.process(job.handlerInputData, (err, result) => {
98 logger.error('Error in job handler %s.', job.handlerName, { error: err })
99 return this.onJobError(jobHandler, job, result, callback)
102 return this.onJobSuccess(jobHandler, job, result, callback)
107 private onJobError (jobHandler, job, jobResult, callback) {
108 job.state = JOB_STATES.ERROR
110 job.save().asCallback(err => {
111 if (err) return this.cannotSaveJobError(err, callback)
113 return jobHandler.onError(err, job.id, jobResult, callback)
117 private onJobSuccess (jobHandler, job, jobResult, callback) {
118 job.state = JOB_STATES.SUCCESS
120 job.save().asCallback(err => {
121 if (err) return this.cannotSaveJobError(err, callback)
123 return jobHandler.onSuccess(err, job.id, jobResult, callback)
127 private cannotSaveJobError (err, callback) {
128 logger.error('Cannot save new job state.', { error: err })
133 // ---------------------------------------------------------------------------