yajob

Job queue on MongoDB. It uses atomic writes to grab jobs and exposes generator friendly API.

Requirements:

• NodeJS >= 4
• MongoDB >= 2.6

Usage

const yajob = require('yajob');
const mails = yajob('localhost/queuedb')
    .tag('mails');

mails.put({
    from: '[email protected]',
    to: '[email protected]',
    body: 'Wow!'
});
// => Promise

for (var mail of yield mails.take(100)) {
    yield sendmail(mail);
}

Processed jobs removed from queue, when for-loop is ended or broken (either with break or exception).

Skip jobs

In some cases you will need to skip taken job. To do this pass into generator false value:

const jobs = yield mails.take(100);
const job = jobs.next().value;

if (value === 'Ohnoez') {
    job.next(false); // Returns Ohnoez back to queue and get next job
}

Priorities

By default, all jobs have priority 0. You can specify sort for queue, in which jobs will be taken:

const important = queue.tag('mail').sort({priority: -1});

API

Yajob(uri, [options])

Returns instance of queue, that stores data in MongoDB.

uri

Type: String

MongoDB URI string.

options

Type: Object

MongoDB MongoClient.connect options.

Methods

put(attrs, [options])

Add job to queue. Returns Promise.

attrs

Type: Object / Array

Data, that will be attached to job. If attrs is an Array - then every Object in attrs considered as new job.

putUnique(attr, [options])

Add unique by attr job to queue. Returns Promise. If job isn't unique promise will be resolved with {result: {ok: 0, n: 0}}

attr

Type: Object

Data, that will be attached to job.

options

Type: Object

• schedule - Date, when job should be available to take
• priority - Number, that represents priority of job

take([count])

Returns Promise that resolves to a Generator, that will emit jobs one by one.

After all jobs are taken from batch - they are considered done and removed from queue.

count

Type: Number
Default: 1

Maximum number of jobs to take from one batch request.

remove(attrs)

Removes jobs, that match attrs from queue. Returns Promise.

close([force])

Closes connections to MongoDB.

Setters

tag(name)

Default: default

Sets name of the MongoDB collection, that will be used to save and get jobs.

delay(milliseconds)

Sets delay for job, that is not scheduled. That is - every job without schedule options will be scheduled on Date() + delay. If job is failed delay will be used to define new shedule on Date() + delay.

trys(number)

Default: Infinity

Sets maximum job trys, before failed status will be assigned. Delay between trys is set by delay(milliseconds) method.

sort(order)

Sets sort order rule for take. Use this, when you need to get jobs, sorted by priority.

Job status

• 0 - new job, that was just added to queue
• 1 - taken job, that was assigned to takenBy worker
• 2 - failed job, that has more attemts than allowed

License

MIT © Vsevolod Strukchinsky