Ghost/ghost/job-manager/lib/job-manager.js

154 lines
5.6 KiB
JavaScript
Raw Normal View History

const path = require('path');
const fastq = require('fastq');
const later = require('@breejs/later');
const Bree = require('bree');
const pWaitFor = require('p-wait-for');
const {UnhandledJobError, IncorrectUsageError} = require('@tryghost/errors');
const logging = require('@tryghost/logging');
const isCronExpression = require('./is-cron-expression');
const assembleBreeJob = require('./assemble-bree-job');
const worker = async (task, callback) => {
try {
let result = await task();
callback(null, result);
} catch (error) {
callback(error);
}
};
const handler = (error, result) => {
if (error) {
// TODO: this handler should not be throwing as this blocks the queue
// throw error;
}
// Can potentially standardise the result here
return result;
};
class JobManager {
/**
* @param {Object} options
* @param {Function} [options.errorHandler] - custom job error handler
* @param {Function} [options.workerMessageHandler] - custom message handler coming from workers
*/
constructor({errorHandler, workerMessageHandler}) {
this.queue = fastq(this, worker, 1);
this.bree = new Bree({
root: false, // set this to `false` to prevent requiring a root directory of jobs
2021-02-22 03:11:11 +03:00
hasSeconds: true, // precision is needed to avoid task overlaps after immediate execution
outputWorkerMetadata: true,
logger: logging,
errorHandler: errorHandler,
workerMessageHandler: workerMessageHandler
});
}
2020-11-05 06:41:16 +03:00
/**
* By default schedules an "offloaded" job. If `offloaded: true` parameter is set,
* puts an "inline" immediate job into the queue.
2020-11-05 06:41:16 +03:00
*
* @param {Object} GhostJob - job options
* @prop {Function | String} GhostJob.job - function or path to a module defining a job
* @prop {String} [GhostJob.name] - unique job name, if not provided takes function name or job script filename
* @prop {String | Date} [GhostJob.at] - Date, cron or human readable schedule format. Manage will do immediate execution if not specified. Not supported for "inline" jobs
* @prop {Object} [GhostJob.data] - data to be passed into the job
* @prop {Boolean} [GhostJob.offloaded] - creates an "offloaded" job running in a worker thread by default. If set to "false" runs an "inline" job on the same event loop
2020-11-05 06:41:16 +03:00
*/
addJob({name, at, job, data, offloaded = true}) {
if (offloaded) {
logging.info('Adding offloaded job to the queue');
let schedule;
if (!name) {
if (typeof job === 'string') {
name = path.parse(job).name;
} else {
throw new IncorrectUsageError({
message: 'Name parameter should be present if job is a function'
});
}
}
if (at && !(at instanceof Date)) {
if (isCronExpression(at)) {
schedule = later.parse.cron(at, true);
} else {
schedule = later.parse.text(at);
}
if ((schedule.error && schedule.error !== -1) || schedule.schedules.length === 0) {
throw new IncorrectUsageError({
message: 'Invalid schedule format'
});
}
logging.info(`Scheduling job ${name} at ${at}. Next run on: ${later.schedule(schedule).next()}`);
} else if (at !== undefined) {
logging.info(`Scheduling job ${name} at ${at}`);
} else {
logging.info(`Scheduling job ${name} to run immediately`);
}
const breeJob = assembleBreeJob(at, job, data, name);
this.bree.add(breeJob);
return this.bree.start(name);
} else {
logging.info('Adding one off inline job to the queue');
this.queue.push(async () => {
try {
if (typeof job === 'function') {
await job(data);
} else {
await require(job)(data);
}
} catch (err) {
// NOTE: each job should be written in a safe way and handle all errors internally
// if the error is caught here jobs implementaton should be changed
logging.error(new UnhandledJobError({
context: (typeof job === 'function') ? 'function' : job,
err
}));
throw err;
}
}, handler);
}
}
/**
* Removes an "offloaded" job from scheduled jobs queue.
* It's NOT yet possible to remove "inline" jobs (will be possible when scheduling is added https://github.com/breejs/bree/issues/68).
* The method will throw an Error if job with provided name does not exist.
*
* NOTE: current implementation does not guarante running job termination
* for details see https://github.com/breejs/bree/pull/64
*
* @param {String} name - job name
*/
async removeJob(name) {
await this.bree.remove(name);
}
/**
* @param {import('p-wait-for').Options} [options]
*/
async shutdown(options) {
await this.bree.stop();
if (this.queue.idle()) {
return;
}
logging.warn('Waiting for busy job queue');
await pWaitFor(() => this.queue.idle() === true, options);
logging.warn('Job queue finished');
}
}
module.exports = JobManager;