flitter-jobs

Flitter-Jobs is a canonical wrapper for BullMQ that provides:

• Simple command to start worker nodes
• Canonical, injectable job classes
• Queue and worker node configurations
• Injectable jobs service

Installation

yarn add flitter-jobs

Now, add the Jobs unit to Units.flitter.js (and the Redis unit, if you don't already have that) to the "Pre-Routing Custom Units" section:

'Redis'         : require('./redis/src/RedisUnit'),
'Jobs'          : require('./jobs/src/JobsUnit'),

Deploy & Config

Run the deployments:

./flitter deploy redis      # if you don't have a redis config
./flitter deploy jobs

Be sure to customize the redis.config.js if you need to. Now, you can edit the jobs.config.js to configure job queues, and worker groups.

Job Queues

A job queue can process any different type of job. It's particularly useful as an organizational device, and a means of managing the relative priority of different queues.

Queues can be processed by different (or multiple) workers.

Worker Groups

A worker group is a set of queues that a particular type of worker will process. This can be a single or multiple queues. You can also have multiple worker processes of the same group type, and they will both process all the queues in the worker group.

Basic Usage

Create a new job definition

./flitter new job Test

Add the code to process the job in the execute method of the new Job class that was created in app/jobs/Test.job.js:

const { Job } = require('flitter-jobs')

class TestJob extends Job {
    static get services() {
        return [...super.services, 'output']
    }

    async execute(job) {
        const { data } = job
        this.output.info('Test job executed with data: ')
        this.output.info(data)
    }
}

module.exports = exports = TestJob

The execute method is async, so it returns a promise. The job is considered complete when this promise resolves. The first parameter of this method is an instance of BullMQ's Job Class.

Create a new queue and worker group

In config/jobs.config.js, create a new queue called low_priority, and a worker group called main to process it:

const jobs_config = {
    queues: [ 'low_priority' ],
    workers: {
        main: {
            queues: ['low_priority'],
        },
    },
}

module.exports = exports = jobs_config

Start the Worker

As a separate process, you can start the worker to process the jobs by starting it with the command:

./flitter worker main

Add Jobs to the Queue

Now, from your main application (or the Flitter shell), you can add jobs to the queue and they will be processed by the worker node:

(flitter)> const lp_queue = _services.jobs.queue('low_priority')
(flitter)> lp_queue.add('Test', { some: 'data' })

In this case, we get the low_priority queue instance and add a new Test job. You can add jobs by their canonical name as they were created.

If you look at the output of the worker process, you should see the processing message:

[INFO] [Test] Test job executed with data:
[INFO] [Test] { some: 'data' }

See the BullMQ docs for more usage information.