server/lib/jobs/job-scheduler.ts

   1 import { forever, queue } from 'async'
   2 import * as Sequelize from 'sequelize'
   3
   4 import { database as db } from '../../initializers/database'
   5 import {
   6   JOBS_FETCHING_INTERVAL,
   7   JOBS_FETCH_LIMIT_PER_CYCLE,
   8   JOB_STATES
   9 } from '../../initializers'
  10 import { logger } from '../../helpers'
  11 import { JobInstance } from '../../models'
  12 import { JobHandler, jobHandlers } from './handlers'
  13
  14 type JobQueueCallback = (err: Error) => void
  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
  31     const jobsQueue = queue<JobInstance, JobQueueCallback>(this.processJob.bind(this))
  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
  65   createJob (transaction: Sequelize.Transaction, handlerName: string, handlerInputData: object, callback: (err: Error) => void) {
  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
  76   private enqueueJobs (err: Error, jobsQueue: AsyncQueue<JobInstance>, jobs: JobInstance[]) {
  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
  86   private processJob (job: JobInstance, callback: (err: Error) => void) {
  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) {
  96         logger.error('Unknown job handler for job %s.', job.handlerName)
  97         return callback(null)
  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
 111   private onJobError (jobHandler: JobHandler<any>, job: JobInstance, jobResult: any, callback: (err: Error) => void) {
 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
 121   private onJobSuccess (jobHandler: JobHandler<any>, job: JobInstance, jobResult: any, callback: (err: Error) => void) {
 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
 131   private cannotSaveJobError (err: Error, callback: (err: Error) => void) {
 132     logger.error('Cannot save new job state.', { error: err })
 133     return callback(err)
 134   }
 135 }
 136
 137 // ---------------------------------------------------------------------------
 138
 139 export {
 140   JobScheduler
 141 }