-import { forever, queue } from 'async'
+import { AsyncQueue, forever, queue } from 'async'
+import * as Sequelize from 'sequelize'
import { database as db } from '../../initializers/database'
import {
JOB_STATES
} from '../../initializers'
import { logger } from '../../helpers'
-import { jobHandlers } from './handlers'
+import { JobInstance } from '../../models'
+import { JobHandler, jobHandlers } from './handlers'
+
+type JobQueueCallback = (err: Error) => void
class JobScheduler {
logger.info('Jobs scheduler activated.')
- const jobsQueue = queue(this.processJob.bind(this))
+ const jobsQueue = queue<JobInstance, JobQueueCallback>(this.processJob.bind(this))
// Finish processing jobs from a previous start
const state = JOB_STATES.PROCESSING
- db.Job.listWithLimit(limit, state, (err, jobs) => {
- this.enqueueJobs(err, jobsQueue, jobs)
-
- forever(
- next => {
- if (jobsQueue.length() !== 0) {
- // Finish processing the queue first
- return setTimeout(next, JOBS_FETCHING_INTERVAL)
- }
-
- const state = JOB_STATES.PENDING
- db.Job.listWithLimit(limit, state, (err, jobs) => {
- if (err) {
- logger.error('Cannot list pending jobs.', { error: err })
- } else {
- jobs.forEach(job => {
- jobsQueue.push(job)
- })
+ db.Job.listWithLimit(limit, state)
+ .then(jobs => {
+ this.enqueueJobs(jobsQueue, jobs)
+
+ forever(
+ next => {
+ if (jobsQueue.length() !== 0) {
+ // Finish processing the queue first
+ return setTimeout(next, JOBS_FETCHING_INTERVAL)
}
- // Optimization: we could use "drain" from queue object
- return setTimeout(next, JOBS_FETCHING_INTERVAL)
- })
- },
+ const state = JOB_STATES.PENDING
+ db.Job.listWithLimit(limit, state)
+ .then(jobs => {
+ this.enqueueJobs(jobsQueue, jobs)
- err => { logger.error('Error in job scheduler queue.', { error: err }) }
- )
- })
+ // Optimization: we could use "drain" from queue object
+ return setTimeout(next, JOBS_FETCHING_INTERVAL)
+ })
+ .catch(err => logger.error('Cannot list pending jobs.', err))
+ },
+
+ err => logger.error('Error in job scheduler queue.', err)
+ )
+ })
+ .catch(err => logger.error('Cannot list pending jobs.', err))
}
- createJob (transaction, handlerName: string, handlerInputData: object, callback) {
+ createJob (transaction: Sequelize.Transaction, handlerName: string, handlerInputData: object) {
const createQuery = {
state: JOB_STATES.PENDING,
handlerName,
}
const options = { transaction }
- db.Job.create(createQuery, options).asCallback(callback)
+ return db.Job.create(createQuery, options)
}
- private enqueueJobs (err, jobsQueue, jobs) {
- if (err) {
- logger.error('Cannot list pending jobs.', { error: err })
- } else {
- jobs.forEach(job => {
- jobsQueue.push(job)
- })
- }
+ private enqueueJobs (jobsQueue: AsyncQueue<JobInstance>, jobs: JobInstance[]) {
+ jobs.forEach(job => jobsQueue.push(job))
}
- private processJob (job, callback) {
+ private processJob (job: JobInstance, callback: (err: Error) => void) {
const jobHandler = jobHandlers[job.handlerName]
+ if (jobHandler === undefined) {
+ logger.error('Unknown job handler for job %s.', job.handlerName)
+ return callback(null)
+ }
logger.info('Processing job %d with handler %s.', job.id, job.handlerName)
job.state = JOB_STATES.PROCESSING
- job.save().asCallback(err => {
- if (err) return this.cannotSaveJobError(err, callback)
-
- if (jobHandler === undefined) {
- logger.error('Unknown job handler for job %s.', jobHandler.handlerName)
- return callback()
- }
-
- return jobHandler.process(job.handlerInputData, (err, result) => {
- if (err) {
- logger.error('Error in job handler %s.', job.handlerName, { error: err })
- return this.onJobError(jobHandler, job, result, callback)
- }
+ return job.save()
+ .then(() => {
+ return jobHandler.process(job.handlerInputData)
+ })
+ .then(
+ result => {
+ return this.onJobSuccess(jobHandler, job, result)
+ },
- return this.onJobSuccess(jobHandler, job, result, callback)
+ err => {
+ logger.error('Error in job handler %s.', job.handlerName, err)
+ return this.onJobError(jobHandler, job, err)
+ }
+ )
+ .then(() => callback(null))
+ .catch(err => {
+ this.cannotSaveJobError(err)
+ return callback(err)
})
- })
}
- private onJobError (jobHandler, job, jobResult, callback) {
+ private onJobError (jobHandler: JobHandler<any>, job: JobInstance, err: Error) {
job.state = JOB_STATES.ERROR
- job.save().asCallback(err => {
- if (err) return this.cannotSaveJobError(err, callback)
-
- return jobHandler.onError(err, job.id, jobResult, callback)
- })
+ return job.save()
+ .then(() => jobHandler.onError(err, job.id))
+ .catch(err => this.cannotSaveJobError(err))
}
- private onJobSuccess (jobHandler, job, jobResult, callback) {
+ private onJobSuccess (jobHandler: JobHandler<any>, job: JobInstance, jobResult: any) {
job.state = JOB_STATES.SUCCESS
- job.save().asCallback(err => {
- if (err) return this.cannotSaveJobError(err, callback)
-
- return jobHandler.onSuccess(err, job.id, jobResult, callback)
- })
+ return job.save()
+ .then(() => jobHandler.onSuccess(job.id, jobResult))
+ .catch(err => this.cannotSaveJobError(err))
}
- private cannotSaveJobError (err, callback) {
- logger.error('Cannot save new job state.', { error: err })
- return callback(err)
+ private cannotSaveJobError (err: Error) {
+ logger.error('Cannot save new job state.', err)
}
}