mini-queue

Job queue

If you have different needs regarding the functionality, please add a feature request.

Installation

npm install --save mini-queue

Usage

 QueueJob State Diagram (methods of JobQueue object)
 ======================

             |
   createJob |
      +------V-----+
      | new        |
      |            |
      +---+--+--+--+
          |  |  | _rejectJob                     +------------+
          |  |  +--------------------------------> reject     |
          |  |                                   |            |
_startJob |  | _queueJob                         +------------+
          |  +------------+
          |               |
          |         +-----v------+ _cancelJob
          |         | queue      +-----------------+
          |         |            |                 |
          |         +---+----^---+                 |
          | _dequeueJob |    |                     |
          |             |    |                     |
          |             |    | _queueJob           |
          |         +---v----+---+               +-V----------+
          |         | dequeue    +---------------> cancel     |
          |         |            | _cancelJob    |            |
          |         +-----+------+               +------------+
          |     _startJob | 
          |               |
          |    +----------+
          |    |
      +---v----v---+  _terminateJob       +------------+
      | process    +----------------------> terminate  |
      |            |                      |(not implem)|
      +-----+------+                      +------------+
            |
            |
            |
      +-----v------+
      | complete   |
      |            |
      +------------+

job.journalEntry for each state (queue, dequeue, process, complete, reject, cancel) stores the time when transition to state occured. If several changes has occured, only the last time is stored(id is job identifier job.id):

       { id: 4,
          new: 2017-11-10T08:37:17.428Z,
          queue: 2017-11-10T08:37:17.428Z,
          dequeue: 2017-11-10T08:37:17.941Z,
          process: 2017-11-10T08:37:17.941Z,
          complete: 2017-11-10T08:37:18.943Z },

For each group and name as provided in option for createJob(), journalEntries are kept ar array in queue.journal (newest is the first, oldest is the last).

Example (group and name not set, default value is used):

journal: { group: 
   { name: 
      [ { id: 5,
          new: 2017-11-10T08:37:17.929Z,
          reject: 2017-11-10T08:37:17.929Z },
. . .
        { id: 2,
          new: 2017-11-10T08:37:16.426Z,
          queue: 2017-11-10T08:37:16.427Z,
          dequeue: 2017-11-10T08:37:16.937Z,
          process: 2017-11-10T08:37:16.937Z,
          complete: 2017-11-10T08:37:17.941Z } ] } } +2s

Up to maxJournalLength option for createJob() records are kept.

Example

You may find this example in demo subdirectory of the package.

"use strict";

process.env.DEBUG = 'queue,app';// + (process.env.DEBUG || '');
var util   = require('util');
var debug  = require('debug')('app');


//var Queue = require('express-queue');
var Queue = require('../');
var queue = new Queue({ activeLimit: 1, queuedLimit: 1, maxJournalLength: 4 });

// create jobs
var maxCount = 5,
    count = 0;

var interval = setInterval(function() {
  var jobData = {};
  // Create new job for the queue
  // If number of active job is less than `activeLimit`, the job will be started on Node's next tick.
  // Otherwise it will be queued.
  var job = queue.createJob(
    jobData, // we may pass some data to job when calling queue.createJob() function
    { group: 'group', name: 'name' } // group/name to be used for journal
  );

  if (++count >= maxCount) {
    clearInterval(interval);

    setTimeout(()=> { // after last job has finished
      debug('journal:', util.inspect(queue.journal, {depth:3}));
      }, 1500); 
  }
}, 500);


// execute jobs

queue.on('process', function(job, jobDone) {
  debug(`queue.on('process'): [${job.id}]: status: ${job.status}, journalEntry: ${JSON.stringify(job.journalEntry)}`);
  // Here the job starts
  //
  // It is also possible to do the processing inside job.on('process'), just be careful
  // to call jobDone() callback once and only once.
  //
  // Value of job.data is set to value passed to queue.createJob()
  //
  // Imitate job processing which takes 1 second to be finished
  setTimeout(function() {
    // Call the callback to signal to the queue that the job has finished
    // and the next one may be started
    jobDone();
    // Now on Node's next tick the next job (if any) will be started
  }, 1000);
});

// Signal about jobs rejected due to queueLimit

queue.on('reject', function(job) {
  debug(`queue.on('reject'): [${job.id}]: status: ${job.status}, journalEntry: ${JSON.stringify(job.journalEntry)}`);
});

Credits

Alexander

Links to package pages:

github.com   npmjs.com   travis-ci.org   coveralls.io   inch-ci.org

License

MIT