tress
v1.1.4
Published
Easy to use asynchronous job queue. Successor of 'caolan/async.queue'.
Downloads
391
Readme
tress
Easy to use asynchronous job queue.
It stores jobs in the memory and runs it asynchronously in parallel with a given concurrency. Successor of caolan/async.queue.
Basically tress
is a clone of queue
from famous caolan/async
but without all other implements of that Swiss Army Knife of asynchronous code. Although tress
was intended to be an improved, better maintained and more safe alternative to caolan/async.queue
, but first and foremost tress
is backward compatible. It means, that everywhere you use async.queue
(except undocumented features) you can write tress
instead and it must work.
You can do like this:
// old code:
var async = require('async');
var q = async.queue(function(job, done){/*...*/});
/*...*/
// new code:
var tress = require('tress');
var q = tress(function(job, done){/*...*/});
/*...*/
Every code using caolan/async.queue
must work with tress
. If it does not work exactly the same way, please start the issue.
All documentation of caolan/async.queue
is right for tress
, but it doesn't describe it completely. Any way, you can use tress
only with this documentation (instead of referencs below) and don't even think about any extra features. Only exception - tress
require node >=8
and doesn't work in browsers.
Main difference between tress
and caolan/async.queue
is that in tress
job not disappear after worker finished. It moves to failed
or finished
(depends of done
first argument) and can be used later. Second difference is that in tress
fields of queue object are more safe. They are readable/writable only in correct way. Also tress
has some new fields in queue object (see in reference).
Install
npm install tress
Synopsis
var tress = require('tress');
// create a queue object with worker and concurrency 2
var q = tress(function(job, done){
console.log('hello ' + job.name);
someAsyncFunction(job, function(err, data){
if (err) {
done(err, 'some message');
} else {
done(null, 'anything');
}
});
}, 2);
// assign a callbacks
q.drain = function(){
console.log('Finished');
};
q.error = function(err) {
console.log('Job ' + this + ' failed with error ' + err);
};
q.success = function(data) {
console.log('Job ' + this + ' successfully finished. Result is ' + data);
}
// add some items to the queue
q.push({name: 'Bob'});
q.push({name: 'Alice'});
// add some items to the queue (batch-wise)
// and specify callback only for that items (not for all queue)
q.push([{name: 'Good'}, {name: 'Bad'}, {name: 'Ugly'}], function (err) {
console.log('finished processing item');
});
// add some items to the front of the queue
q.unshift({name: 'Cristobal Jose Junta'});
Reference
tress(worker, [concurrency])
creates queue object that will store jobs and process them with worker
function in parallel (up to the concurrency
limit).
Arguments:
worker(job, done)
- An asynchronous function for processing a queued data.job
- data for processing. It may be primitive or object.done(err, ...args)
- callback function.worker
must calldone
when finished.done
may take various argumens, but first argument must be error (if job failed), null/undefined (if job successfully finished) or boolean (if job need to be returned to queue head (iftrue
) or to queue tail (iffalse
) for retry).
concurrency
- An integer for determining how many worker functions should be run in parallel. Defaults to 1. If negative - no parallel and delay between worker calls (concurrency -1000 sets 1 second delay).
Queue object properties
started
- stillfalse
till any items have been pushed and processed by the queue. Than becametrue
and never change in queue lifecycle (Not writable).concurrency
- This property for alter the concurrency/delay on-the-fly.buffer
A minimum threshold buffer in order to say that the queue is unsaturated. Defaults toconcurrency / 4
.paused
- a boolean for determining whether the queue is in a paused state. Not writable (usepause()
andresume()
instead).waiting
(new) - array of queued jobs.active
(new) - array of jobs currently being processed.failed
(new) - array of failed jobs (done
callback was called from worker with error in first argument).finished
(new) - array of correctly finished jobs (done
callback was called from worker withnull
orundefined
in first argument).
Note, that properties waiting
, active
, failed
and finished
are not writable, but they point to arrays, that you can cahge manually. Do it carefully.
Queue object methods
Note, that in tress
you can't rewrite methods.
push(job, [callback])
orunshift(job, [callback])
- add a new job to the queue.push
adds job to queue tail andunshift
to queue head.job
- Job to add to queue. Instead of a single job, array of jobs can be submitted.callback(err, ...args)
- optional callback.tress
calls this callback once the worker has finished processing the job. With same arguments as indone
callback.
pause()
- a function that pauses the processing of new jobs untilresume()
is called.resume()
- a function that resumes the processing of queued jobs paused bypause
.kill()
- a function that removes the drain callback and empties queued jobs.remove()
a function that removes job from queue.length()
- a function returning the number of items waiting to be processed.running()
- a function returning the number of items currently being processed.workersList()
- a function returning the array of items currently being processed.idle()
- a function returning false if there are items waiting or being processed, or true if not.save(callback)
(new) - a function that runs a callback with object, that contains arrays ofwaiting
,failed
, andfinished
jobs. If there are anyactive
jobs at the moment, they will be concatenated towaiting
array.load(data)
(new) - a function that loads new arrays fromdata
object towaiting
,failed
, andfinished
arrays and setsactive
to empty array. Rise error ifstarted
istrue
.status(job)
(new) a function returning the status ofjob
('waiting'
,'running'
,'finished'
,'pending'
or'missing'
).
Queue objects callbacks
You can assign callback function to this six properties. Note, you can't call any of that calbacks manually.
saturated()
- a callback that is called when the number of running workers hits the concurrency limit, and further jobs will be queued.unsaturated()
- a callback that is called when the number of running workers is less than the concurrency & buffer limits, and further jobs will not be queued.empty()
- a callback that is called when the last item from the queue is given to a worker.drain()
- a callback that is called when the last item from the queue has been returned from the worker.error(...args)
(new) - a callback that is called when job failed (worker calldone
with error as first argument).success(err, ...args)
(new) - a callback that is called when job correctly finished (worker calldone
withnull
orundefined
as first argument).retry(err, ...args)
(new) - a callback that is called when job returned to queue for retry (worker calldone
with boolean as first argument).
Job lifecycle
After pushing/unshifting to queue every job passes through the following steps:
- processed in worker
- sent to
done
(asthis
or explicitly as one of parameters if you prefere arraw-callbacks) - moved from
active
tofailed
/finished
/waiting
(wether first parameter ofdone
is error, null or boolean) - sent to job callback (from
push
/unshift
) - sent to
error
/success
/retry
callback (wether first parameter ofdone
is error, null or boolean)
License
MIT