[!CAUTION] This package is not compatible with AdonisJS v5.

• Introduction
• Installation
• Folders
• Job Usage
  • Basic
  • Batch enqueue
  • Delayed enqueue
  • Repeated Enqueue
• Handle Failure Jobs
  • Job.onFailure
  • Failure Event
• Demonstration
  • Send Mail Job
• Plugin
  • Plugin.jobLock
  • Plugin.queueLock
  • Plugin.delayQueueLock
  • Plugin.retry
  • Plugin.noop
  • Custom Your Own Plugin
• Configuration
• Concepts & Components
  • Queue
    • Example: Getting a total count of pending jobs
  • Worker
    • Multi Worker
  • Scheduler
• Deployment
  • Development
  • Production
• Web UI
• Notice 1: for the graceful exit
• Notice 2: Job does not @injectable
• Who's Using
• Contributors
• Reference
• License

Introduction

@fellas/adonisjs-resque add the queue ability to adonis base on node-resque. Resque is a background job system backed by Redis.

Installation

1node ace add @fellas/adonisjs-resque

[!IMPORTANT] @adonisjs/redis is required for resque redis connection.

Folders

Jobs are placed in folder app/jobs by default. You can import job through sub-path.

1import Example from '#jobs/example'

[!TIP] Please follow this instruction: The sub-path imports.

Both package.json and tsconfig.json are required to add the job path:

• add to package.json

1"#jobs/*": "./app/jobs/*.js"

• add to field compilerOptions.paths in tsconfig.json

1"#jobs/*": ["./app/jobs/*.js"]

Job Usage

You can create a resque job by adonis command: node ace make:job <YourJobName>

Basic

Create a basic example job: node ace make:job BasicExample. Every job has a perform method. It runs in the background, which consumer from the node-resque queue.

1import { BaseJob } from '@fellas/adonisjs-resque'
2export default class BasicExample extends BaseJob {
3  async perform(name: string) {
4    this.logger.info(`Hello ${name}`)
5  }
6}

Now you can enqueue this job.

1import BasicExample from '#jobs/basic_example'
2await BasicExample.enqueue('Bob')

Console would print Hello Bob.

[!WARNING] Make sure your Job ClassName is unique in the queue. and the arguments of it's method perform would be serialize to a json string to redis under the hood.

Batch enqueue

1await BasicExample.enqueueAll([
2  ['Alice'],
3  ['Bob'],
4  ['Carol'],
5  ['Dave'],
6  ['Eve']
7])

Delayed enqueue

1const oneSecondLater = 1000
2await BasicExample.enqueue('Bob').in(oneSecondLater)

Or enqueue at a specify timestamp

1const fiveSecondsLater = new Date().getTime() + 5000
2await BasicExample.enqueue('Bob').at(fiveSecondsLater)

Repeated Enqueue

class Job has the schedule properties.

• interval, .e.g 5s, 15m, 2h and 1d. package ms for more details.
• cron, for cron syntax, look up the croner package

The example below enqueue in both every 1 second and 5 minutes, since it's cron/interval settings.

1export default class BasicExample extends BaseJob {
2  // enqueue job cronly
3  cron: '*/1 * * * * *',
4  // enqueue every five minutes
5  interval: '5m',
6  async perform(name: string) {
7    this.logger.log(`Hello ${name}`)
8  }
9}

Handle Failure Jobs

Job.onFailure

You can handle failure jobs by defining a onFailure method. Once a job fails, it would be called before moving to the failed queue.

1import { BaseJob, type ResqueFailure } from '@fellas/adonisjs-resque'
2class Job extends BaseJob {
3      async onFailure(failure: ResqueFailure) {
4        console.log('resque job failured:', failure)
5    }
6}

The definition of interface ResqueFailure

1export interface ResqueFailure {
2    // Only the failure emitted by MultiWorker has a workerId
3    workerId?: number
4    queue: string
5    job: NodeResqueJob
6    failure: Error
7    duration: number
8}

[!TIP] If you are using retry plugin, the onFailure method will be called only if the job has exceeded the retry limit.

Failure Event

Another way to handle failure jobs is to listen to the resque:failure event.

go start/events.ts to handle.

1import emitter from '@adonisjs/core/services/emitter'
2
3emitter.on('resque:failure', (failure) => {
4    console.error('resque:failure', failure)
5})

Demonstration

Send Mail Job

In the Adonis Documentation, they use bullmq as mail queueing example. But if we want to use @fellas/adonisjs-resque for mail.sendLater, how to do?

1. Create a Mail Job
   Run node ace make:job Mail to create the mail job, then edit it in app/jobs/mail.ts

1import { BaseJob } from '@fellas/adonisjs-resque'
2import mail from '@adonisjs/mail/services/main'
3import { MessageBodyTemplates, NodeMailerMessage } from '@adonisjs/mail/types'
4
5interface Options {
6    mailMessage: {
7        message: NodeMailerMessage;
8        views: MessageBodyTemplates;
9    }
10    config: any
11}
12export default class Mail extends BaseJob {
13    async perform(option: Options) {
14        const { messageId } = await mail.use('smtp')
15            .sendCompiled(option.mailMessage, option.config)
16        this.logger.info(`Email sent, id is ${messageId}`)
17    }
18}

2. Custom mail.setMessenger in a service provider
   You can add the below code snippet to a boot method of any service provider.

1const mail = await this.app.container.make('mail.manager')
2mail.setMessenger(() => {
3  return {
4    async queue(mailMessage, sendConfig) {
5      return Mail.enqueue({ mailMessage, config: sendConfig })
6    }
7  }
8})

3. mail.sendLater is available now! Let's have a try. :shipit:

1await mail.sendLater((message) => {
2  message.to('your-address@example.com', 'Your Name')
3  .subject('Hello from @fellas/adonisjs-resque')
4  .html(`<strong>Congratulations!</strong>`)
5})

[!CAUTION] You should insure @adonisjs/mail has a correct config, you'd better to test it first.

Plugin

@fellas/adonisjs-resque encapsulated the default node-resque plugins in a smooth way, with the better typing support.

The default node-resque plugins are:

• Retry
• JobLock
• Retry
• QueueLock
• DelayQueueLock
• Noop

You can adding them to the plugins property of your job class, by Plugin.<PluginName>({...pluginOptions})

Plugin.jobLock

JobLock plugin is used to prevent a job to be performed (hook in beforePerform), which has the same name, queue and args is already running.

If reEnqueue is true, the job will be put back (re-enqueue) with a delay .

1import { BaseJob, Plugin } from "@fellas/adonisjs-resque"
2
3export default class ExampleJob extends BaseJob {
4    plugins = [
5        Plugin.jobLock({ reEnqueue: false, enqueueTimeout: 3000 })
6    ]
7    async perform() {
8        this.logger.info('Example job started')
9        await sleep(60000)
10        this.logger.info('Example job done')
11    }
12}
13
14function sleep(ms: number) {
15    return new Promise(resolve => setTimeout(resolve, ms));
16}

Plugin.queueLock

Similar to jobLock, but it would be prevented before enqueue. If a job with the same name, queue, and args is already in the queue, do not enqueue it again.

QueueLock Options:

1export interface QueueLockOptions {
2    /**
3     * in seconds
4     */
5    lockTimeout?: number
6    key?: LockKey
7}

Plugin.delayQueueLock

Same as queueLock, but it is for the delay queue.

[!IMPORTANT] How "Prevented Before Enqueue" Works?
The delayQueueLock only prevents immediate enqueue operations; it doesn't prevent delayed enqueue operations, such as those using in or at. In node-resque, only non-delayed enqueue operations trigger the beforeEnqueue hook. If you use in or at, the hook is not triggered. Here is source code

Therefore, in the following code, the second operation won't be enqueued:

1
2export default class ExampleJob extends BaseJob {
3    plugins = [
4        Plugin.delayQueueLock()
5    ]
6    async perform() {}
7}
8await ExampleJob.enqueue().in(1000)
9// won't enqueue to queue
10await ExampleJob.enqueue()

However, in the following code, the job will perform twice, no matter if the delayQueueLock plugin is enabled or not:

1await ExampleJob.enqueue().in(1000)
2await ExampleJob.enqueue().in(1500)

Plugin.retry

If a job fails, retry it times before finally placing it into the failed queue You can specify the retryDelay option to delay, or set the backoffStrategy to the delay ms array. Note: It will retry by key, composed of queueName, jobName and args. Not retry by the enqueuing action.

1import { BaseJob, Plugin } from "@fellas/adonisjs-resque"
2
3export default class ExampleJob extends BaseJob {
4    plugins = [
5        Plugin.retry({
6            retryLimit: 3,
7            backoffStrategy: [1000, 3000, 8000]
8        })
9    ]
10    async perform() {
11        const res = await fetch(`https://dummyjson.com/products/1`)
12        this.logger.info(`Response status is ${res.status}`)
13    }
14}

Plugin.noop

you can customize the error logger by option logger

Custom Your Own Plugin

You can create your own plugin by extending the BasePlugin class, it also implements the Plugin abstract class from node-resque. See more details in node-resque plugins.

1import { BasePlugin } from "@fellas/adonisjs-resque"
2export default class MyPlugin extends BasePlugin {
3    /**
4     * Plugin options here
5     * with default values
6     */
7    options = {
8        foo: 'bar'
9    }
10    async beforeEnqueue() {
11    }
12    async afterEnqueue() {
13    }
14    async beforePerform(){
15    }
16    async afterPerform() {
17    }
18}

Applying it in your job class.

1import { BaseJob } from "@fellas/adonisjs-resque"
2// import MyPlugin from './my-plugin'
3
4class ExampleJob extends BaseJob {
5    plugins = [
6        MyPlugin.create({ foo: 'baz' })
7    ]
8}

Configuration

Here is an example of config/resque.ts

1{
2    /**
3     * redis connection config from @adonisjs/redis
4     */
5    redisConnection: 'main',
6    /**
7     * run web & worker in same process, if enabled
8     * You need to run command node ace resque:start if it is turned off
9     *
10     * it's convenient but NOT Recommanded in production
11     * also, DO NOT enable for math-heavy jobs, even in the dev or staging environment.
12     * 
13     */
14    runWorkerInWebEnv: true,
15    /**
16     * when runScheduler enabled, it starts with worker
17     * if you'd like to run scheduler in the separated processes
18     * please turn runScheduler off, and run command
19     * node ace resque:start --scheduler
20     */
21    runScheduler: true,
22    /**
23     * enable node-resque multiworker
24     * @docs https://github.com/actionhero/node-resque?tab=readme-ov-file#multi-worker
25     */
26    isMultiWorkerEnabled: true,
27    /**
28     * the first argument in MultiWorker constructor
29     */
30    multiWorkerOption: {
31        minTaskProcessors: 1,
32        maxTaskProcessors: 10
33    },
34    /**
35     * the argument for Worker constructor, if multiWorker is not enabled
36     */
37    workerOption: {
38    },
39    /**
40     * the default queue name for jobs to enqueue
41     */
42    queueNameForJobs: 'default',
43    /**
44     * queue name for workers to listen,
45     * is a string or an array of string
46     * setting a proper queue name could change their priorities 
47     * e.g. queueNameForWorkers: "high-priority, medium-priority, low-priority"
48     * All the jobs in high-priority will be worked before any of the jobs in the other queues. 
49     */
50    queueNameForWorkers: '*',
51    queueNameForWorkers: '*',
52    /**
53     * set null to use the default logger
54     */
55    logger: null,
56    // verbose mode for debugging
57    verbose: true
58}

Concepts & Components

Queue

Queue is a redis list under the hood, consumed(redis pop) in worker. Each job has a queueName, defined in Job class, default is default.

1class Example extends BaseJob {
2  queueName = 'default'
3}

You can enqueue jobs into queue from anywhere, .e.g, from a Adonis web controller or from another Job class: Example.enqueue(...).

That means a redis push.

For failed jobs, they have been move to queue failed.

The node-resque queue object is exposed in a container:

1const queue = await app.container.make('queue')

You can interact with it, find API here: https://node-resque.actionherojs.com/classes/Queue.html.

Example: Getting a total count of pending jobs

1const pendingJobsTotal = await queue.length(<YourQueueName>)

Worker

The jobs you created are pulling and executed by workers actually. It is a adonis ace command, long-running under the console environment.

You can start a worker by command:

1node ace resque:start --worker

It has been executed by web server in default, so you don't need to run it.

[!IMPORTANT] In the production environment, we don't recommend you to use the web server to run the worker. Instead, you should run the worker in the separated process, or multiple server.

Every worker has its own workerName, which composed of os.hostname() + ":" + process.pid + "+" + counter;

Multi Worker

MultiWorker comes from node-resque, is a pool manager for workers. It starts multiple workers in the same process, not in child process. Threfore, those workers have a same pid in their workerName.

This is designed for I/O intensive jobs, not CPU. So don't get confused by the word multi.

[!TIP] If you'd like to handle CPU intensive jobs, you can start multipe processes(by running the ace command), even on multipe machines.

Multi worker is enabled by isMultiWorkerEnabled in config by default. If you'd like start a single worker, you can set the config to false, or use the flag --is-multi to control:

1node ace resque:start --worker --is-multi=false

Scheduler

Scheduler is a kind of specialized internal worker. It is a coordinator of the jobs.

the Utilities are:

• Support croner & interval jobs
• Process delayed job (job.in/job.at)
• Checking stuck workers
• General cluster cleanup
• ...

[!TIP]

Scheduler could and should be run in many processes(machines), only one will be elected to be the leader, and actually do work; the others are backup. For more informations, see node-resque leader scheduler: https://github.com/actionhero/node-resque?tab=readme-ov-file#job-schedules

By default, scheduler starts with an worker together. You can modify runScheduler to false in config, to changing this behavior.

To start a stand alone scheduler, you can run

1node ace resque:start --worker=false --schedule

Deployment

Resque is a distributed job queue system, it can be deloyed to separated processes or servers.

Development

The scheduler and multiworker are started with web server together. You don't need to take any additional steps to make it work.

Production

You should disable the config runWorkerInWebEnv, and run node ace resque:start command in separated processes.

It is recommended to start mutiple schedules and workers. For I/O intensive jobs, you should start multiWorker with --is-multi on; for CPU intensive jobs, no benefits from that.

Usually, we build a docker image for both web server and jobs, start worker and scheduler by setting the entrypoint. They could share the same other configures, like environment variables.

docker-compose.yml

1    ...
2    entrypoint: ["node", "ace", "resque:start", "--worker", "--schedule"]

For people who don't use docker, you can also use supervisord or pm2 to manage the processes.

Web UI

node-resque also compatible with some Resque Web UI, .e.g resque-web

Here is docker-compose.yml an example

1services:
2  redis:
3    image: redis
4  resque-web:
5    image: appwrite/resque-web:1.1.0
6    ports:
7      - "5678:5678"
8    environment:
9      - RESQUE_WEB_HOST=redis # (OPTIONAL - Use only if different than the default 127.0.0.1)
10      - RESQUE_WEB_PORT=6379  # (OPTIONAL - Use only if different the default 6379)
11      - RESQUE_WEB_HTTP_BASIC_AUTH_USER= # (OPTIONAL - if not set no password used)
12      - RESQUE_WEB_HTTP_BASIC_AUTH_PASSWORD=  # (OPTIONAL - if not set no password used)
13    depends_on:
14      - redis
15    restart: unless-stopped

if redis server has a password, you can add a entrypoint

1    entrypoint:
2      - resque-web
3      - -FL
4      - -r
5      # Change your redis password here,
6      # default redis user is default
7      - "redis://<redis-user>:<redis-password>@<host>:<port>"
8      - /config.ru

Notice 1: for the graceful exit

resque require the graceful exit, or schedulers would waiting for a leader election.

node-resque may not exit in dev environment if you exit by ctrl+c. you can change bin/server.ts

1app.listenIf(app.managedByPm2, 'SIGINT', () => app.terminate())

to

1app.listen('SIGINT', () => app.terminate())

if adonis dev server terminated but process still there, you can send SIGTERM signal to all node process(on macOS) killall node

You can also check the redis key resque:resque_scheduler_leader_lock, which value is scheduler name contains pid of the leader process. it should be release once server terminated.

Notice 2: Job does not @injectable

Since our job is a Thenable class, that made it not able be inject(related issue), which means @inject decorator won't work.

You can use app.container.make() in job instead.

Who's Using

Create an issue to submit your project.

Contributors

This project is contributed by u301 team for giving back to the AdonisJS community.

Reference

• node-resque

License

MIT