Nxus-worker-queue NPM

nxus-worker-queue

Worker Queue Module

Using Redis for pub/sub background tasks

Installation

    > npm install nxus-worker-queue --save

Configuration Options

    "worker_queue": {
      "redis_url": "redis://localhost:6379",
      "cleanInterval": 3600000
    }

It's conventional to use a configuration variable to set the Redis URL in the production environment. For example:

    let config = {}
    if (process.env.REDIS_URL)
      config.worker_queue = { redis_url: process.env.REDIS_URL }
    application.start(config)

Usage

For each task, you need to define a unique task name.

Register a worker handler

    import {workerQueue} from 'nxus-worker-queue'
    workerQueue.worker('myBackgroundTask', ({data}) => {
      this.log.debug("Hello", data.hi)
    })

Request task processing

    import {workerQueue} from 'nxus-worker-queue'
    let job = workerQueue.task('myBackgroundTask', {hi: world})

The job object and notification of completed tasks

The worker queue module interacts with Redis through the intermediary Bull package. This "fastest, most reliable, Redis-based queue for Node" is "carefully written for rock solid stability and atomicity". For documentation, a good place to start is the Reference page.

The task() method returns a Bull Job object that allows you to interact with the background task.

In particular, the Job object exposes a finished() method that, when invoked, returns a promise that resolves when the job finishes. The value of the promise corresponds to the value of the promise returned by the task handler.

    let job = workerQueue.task('myBackgroundTask', {hi: world})
    job.finished().then((rslt) = { console.log('background task finished: ', rslt) })

API

WorkerQueue

Extends NxusModule

Worker Queue module for background tasks

worker

Provide a task handler

Parameters

taskName string Name of the task (channel) to listen for
handler function Handler for processing task requests; should return a promise that resolves on completion
opts (optional, default {})

Examples

workerQueue.worker('backgroundJob', (msg) -> {})

task

Request handling of a background task

Parameters

taskName string Name of the task (channel) to publish to
message object Options for the task worker; must be JSON serializable
opts (optional, default {})

Examples

workerQueue.task('backgroundJob', {hi: 'world'})

Returns object Bull job object

clean

Cleans the current queue for the given taskName.

Parameters

taskName string The queue/task name to clean.
type String The type of message to clean. Defaults to 'completed'. (optional, default 'completed')
delay Number The grace period. Messages older than this will be cleaned. Defaults to 1 hour. (optional, default 3600000)

cleanAll

Cleans all queues for the specified message type.

Parameters

type String The type of message to clean. Defaults to 'completed'. (optional, default 'completed')
delay Number The grace period. Messages older than this will be cleaned. Defaults to 1 hour. (optional, default 3600000)