Queue

Queue

Job Queue

Constructor

new Queue(name, redisopt, optionsopt)

Source:

Create a new Queue
The redis and enableWatchdog settings are required only the first time to init
Can also be set beforehand by calling Queue.init()

Parameters:
Name Type Attributes Default Description
name string

Name of the queue
redis object <optional>
 {port: 6379, host: '127.0.0.1'}

Redis connection settings object
options boolean | queueOpts <optional>
 {}

enableWatchdog or opts object (default = {})
Read more here : https://github.com/Automattic/kue#unstable-redis-connections

Methods

(async, static) exit(timeoutopt) → {boolean}

Source:

Function shuts down the Queue gracefully.
Waits for active jobs to complete until timeout, then marks them failed.

Parameters:
Name Type Attributes Default Description
timeout number <optional>
 10000

Time in milliseconds, default = 10000
Returns:
Type
boolean

(async, static) getCount(queue, jobType) → {number}

Source:

Return count of jobs in Queue of JobType

Parameters:
Name Type Description
queue string

Queue name
jobType string

One of {'inactive', 'delayed' ,'active', 'complete', 'failed'}
Returns:

count

Type
number

(static) init(redisopt, enableWatchdogopt)

Source:
See:

Initialise the redis connection

Parameters:
Name Type Attributes Default Description
redis object <optional>
 {port: 6379, host: '127.0.0.1'}

Redis connection settings object
enableWatchdog boolean <optional>
 false

Will watch for stuck jobs due to any connection issues

(async, static) processJobById(jobId, processor) → {jobDetails}

Source:

Manualy process a specific Job. Returns existing result if job already processed

Parameters:
Name Type Description
jobId number

Id of the job to be processed
processor processorCallback

Function to be called to process the job data, without ctx
Returns:

Result of processor function and job object of completed job

Type
jobDetails

(async, static) status(jobId) → {jobDetails}

Source:

Function to query the status of a job

Parameters:
Name Type Description
jobId number

Job id for which status info is required
Returns:

Object full of job details like state, time, attempts, etc.

Type
jobDetails

(async) activeJobs() → {number}

Source:

Return count of active jobs in Queue

Returns:

activeCount

Type
number

(async) addAndProcess(input, opts, timeoutopt) → {any}

Source:

Add a job to the Queue, wait for it to process and return result
Preferably set PRIORITY HIGH or it might timeout if lots of other tasks are in queue
Queue will process job only if timeout is not passed when processing begins

Parameters:
Name Type Attributes Default Description
input any

Job data
opts addOpts
timeout number <optional>
 180000

wait for this time else throw err
Returns:

result

Type
any

(async) addJob(input, opts) → {number}

Source:

Add a job to the Queue

Parameters:
Name Type Description
input *

Job data
opts addOpts
Returns:

The ID of the job created

Type
number

(async) addProcessor(processor, concurrencyopt)

Source:

Attach a processor to the Queue which will keep getting jobs as it completes them

Parameters:
Name Type Attributes Default Description
processor processorCallback

Function to be called to process the job data
concurrency number <optional>
 1

The number of jobs this processor can handle parallely

(async) cleanup(olderThanopt)

Source:

Cleanup function to be called during startup,
resets active jobs older than specified time

Parameters:
Name Type Attributes Default Description
olderThan number <optional>
 5000

Time in milliseconds, default = 5000

(async) completedJobs() → {number}

Source:

Return count of completed jobs in Queue
Might return 0 if removeOnComplete was true

Returns:

completeCount

Type
number

(async) delayedJobs() → {number}

Source:

Return count of delayed jobs in Queue

Returns:

delayedCount

Type
number

(async) delete(olderThanopt)

Source:

Removes any old jobs from queue
older than specified time

Parameters:
Name Type Attributes Default Description
olderThan number <optional>
 3600000

Time in milliseconds, default = 3600000 (1 hr)

(async) failedJobs() → {number}

Source:

Return count of failed jobs in Queue

Returns:

failedCount

Type
number

(async) inactiveJobs() → {number}

Source:

Return count of inactive jobs in Queue

Returns:

inactiveCount

Type
number

(async) pauseProcessor(timeoutopt)

Source:

Pause Queue processing
Gives timeout time to all workers to complete their current jobs then stops them

Parameters:
Name Type Attributes Default Description
timeout number <optional>
 5000

Time to complete current jobs in ms

(async) pendingJobs() → {number}

Source:

Alias for inactiveJobs

Returns:

inactiveCount

Type
number

(async) processJob(processor) → {jobDetails}

Source:

Process a single job in the Queue and mark it complete or failed,
for when you want to manually process jobs

Parameters:
Name Type Description
processor processorCallback

Function to be called to process the job data, without ctx
Returns:

Job object of completed job

Type
jobDetails

resumeProcessor()

Source:

Resume Queue processing

setAttempts(attempts)

Source:

Set default number of retry attempts for any job added later

Parameters:
Name Type Description
attempts number

Number of attempts (>= 0), default = 1

setDelay(delay)

Source:

Set delay b/w successive jobs for any job added later

Parameters:
Name Type Description
delay number

Delay b/w jobs, milliseconds, default = 0

setNoFailure(noFailure)

Source:

Sets default noFailure for any job added to this Queue from now on.
This will mark the job complete even if it fails when true

Parameters:
Name Type Description
noFailure boolean

default = false

setRemoveOnCompletion(removeOnComplete)

Source:

Sets default removeOnComplete for any job added to this Queue from now on

Parameters:
Name Type Description
removeOnComplete boolean

default = false

setTTL(ttl)

Source:

Set default TTL (time to live) for new jobs added from now on,
will fail job if not completed in TTL time

Parameters:
Name Type Description
ttl number

Time in milliseconds, infinite when 0. default = 0