Classy Job

A class-based wrapper around job collection, a powerful and easy to use job manager for Meteor.

Adding this package to your Meteor application adds Job and JobsWorker classes into the global scope.

Server side only (with JobsWorker available on the client side as well).

Installation

meteor add peerlibrary:classy-job

Jobs

The basic use is to extend Job class and implement the run method:

1class ExampleJob extends Job {
2  run() {
3    this.logInfo("Hello world from: " + this.data.name);
4  }
5}
6
7ExampleJob.register();

run is expected to be blocking and when it returns the job is seen as successfully completed. The return value is stored in the job's result. If run throws an exception, the job is marked as failed and exception is logged.

Then, when you want to put a new instance of the job into the queue, run:

1new ExampleJob({name: "Foo"}).enqueue({skipIfExisting: true});

Class constructor receives data which is then available to the job when it is run (data is stored in the database and retrieved on a worker, so it should be EJSON-serializable).

enqueue accepts the following options:

• skipIfExisting, if true, the job will not be enqueued if a not-completed job already exists
• skipIncludingCompleted, if true, together with skipIfExisting, the job will not be enqueued if a job already exists, even if completed
• depends, if set it is passed to job collection's depends
• priority, if set it is passed to job collection's priority
• retry, if set it is passed to job collection's retry
• repeat, if set it is passed to job collection's repeat
• delay, if set it is passed to job collection's delay
• after, if set it is passed to job collection's after
• save, if set it is passed to job collection's save

When using skipIfExisting there is a slight race-condition possible. In the worst case there will be some duplicate work done. This should not be a problem because jobs ought to be idempotent anyway.

Initialization

Call JobsWorker.initialize() in your app on both client and server to initialize the worker environment and JobsWorker.collection collection.

Possible options for JobsWorker.initialize with defaults:

1JobsWorker.initialize({
2  collectionName: 'JobQueue',
3  workerInstances: parseInt(process.env.WORKER_INSTANCES || '1'),
4  stalledJobCheckInterval: 60 * 1000, // ms
5  promoteInterval: 15 * 1000 // ms
6});

You can use WORKER_INSTANCES environment variable or workerInstances option to control how many workers are enabled across all Meteor instances for your app. If set to 0 the current Meteor instance will not run a worker.

Call JobsWorker.start on the server to start the worker:

1Meteor.startup(function () {
2  JobsWorker.start()
3});

JobsWorker.start call will not do anything if workerInstances is 0. Alternatively, you can simply do not call JobsWorker.start.

Starting is randomly delayed a bit to distribute the behavior of workers equally inside configured intervals.

Jobs are executed serially inside a given worker, one by one.

Working with jobs

Job classes will be instantiated automatically every time they are ready and their run method will be called. Inside your run method you can call other methods, for example log or progress to report on job's progress.

You can use Job.find(query, fields) and Job.findOne(query, fields) to get instances of jobs from the database. Resulting objects will be proper instances of your job classes of the correct type for each result object. If you want only types of a particular class you can limit the query yourself:

1Job.find({
2  type: ExampleJob.type()
3});

If you need access to any other functionality of job collection not available directly through existing methods, you can call getQueueJob to get underlying job collection's job. We call job collection's jobs queue jobs.

So, for example, to cancel a job, you can do:

1Job.findOne({
2  'data.argument': 'foobar'
3}).getQueueJob().cancel();

You can use JobsWorker.collection to access underlying job collection. To convert its jobs (queue jobs) to an instance of a class-based job you can use Job.fromQueueJob(job).