rethinkdb-job-queue - npm Package Compare versions

Comparing version 0.1.2 to 0.2.0

CHANGELOG.md

 # `rethinkdb-job-queue` Change log
 
+## v0.2.0 / 2016-09-27
+
+*   Extended failed job `dateCreated` value in queue-process tests.
+*   Added worker stops processing on global `cancelled` event.
+*   Refactored `Queue.process`.
+
 ## v0.1.2 / 2016-09-20
@@ -10,14 +16,14 @@
 
-*   Added the 'job-update' module.
-*   Added 'job.update()' method.
-*   Changed 'added' job status to 'waiting'.
+*   Added the `job-update` module.
+*   Added `job.update()` method.
+*   Changed `added` job status to `waiting`.
 
 ## v0.1.0 / 2016-09-16
 
-*   Added 'db-driver' module for connection testing.
-*   Changed 'Queue()' constructor API to add connection options.
+*   Added `db-driver` module for connection testing.
+*   Changed `Queue()` constructor API to add connection options.
 *   Standardized time related options to milliseconds.
-*   Added 'datetime' module to add and format dates.
-*   Updated 'is' module to include dateBefore, dateAfter, dateBetween.
-*   Removed the 'moment' library.
+*   Added `datetime` module to add and format dates.
+*   Updated `is` module to include `dateBefore`, `dateAfter`, `dateBetween`.
+*   Removed the `moment` library.
 
@@ -29,7 +35,7 @@ ## v0.0.8 / 2016-09-09
 *   Updated `is.date()` to remove `moment.isDate()`.
-*   Added 'Delayed Job' tests to queue-process tests.
+*   Added `Delayed Job` tests to queue-process tests.
 
 ## v0.0.7 / 2016-08-23
 
-*   Fixed next() calls.
+*   Fixed `next()` calls.
 *   Minor refactor.
@@ -40,4 +46,4 @@ *   README updated.
 
-*   Changed the next() function signature.
-*   Added catch() to next() Promise.
+*   Changed the `next()` function signature.
+*   Added `catch()` to `next()` Promise.
 *   README updated.
@@ -47,22 +53,22 @@
 
-*   Changed return values of Queue.removeJob to job ids.
-*   Removed the Queue.connection alias property.
-*   Added the 'pausing' Queue event.
-*   Updated examples to work with the next() Promise.
-*   Fixed a require path in queue-add-job.
+*   Changed return values of `Queue.removeJob` to job ids.
+*   Removed the `Queue.connection` alias property.
+*   Added the `pausing` Queue event.
+*   Updated examples to work with the `next()` Promise.
+*   Fixed a require path in `queue-add-job`.
 
 ## v0.0.4 / 2016-08-18
 
-*   Removed 'retry' job priority.
+*   Removed `retry` job priority.
 *   Added process timeout extender when progress updated.
-*   Added priorityAsString to 'Job.getCleanCopy'.
-*   Fixed job-options so it keeps current if only one supplied.
-*   Removed Job.setPayload.
-*   Switched to using Map() for queue-process job timeouts.
+*   Added `priorityAsString` to `Job.getCleanCopy`.
+*   Fixed `job-options` so it keeps current if only one supplied.
+*   Removed `Job.setPayload`.
+*   Switched to using `Map()` for queue-process job timeouts.
 
 ## v0.0.3 / 2016-08-12
 
-*   Removed 'master' Queue options, only using 'masterInterval'.
-*   masterInterval default options changed from 300 sec to 310.
-*   Internal parameter rename from "job" or "jobId" to "jobOrId".
+*   Removed `master` Queue options, only using `masterInterval`.
+*   `masterInterval` default options changed from 300 sec to 310.
+*   Internal parameter rename from `job` or `jobId` to `jobOrId`.
 *   Updated README with better examples.
@@ -72,4 +78,4 @@
 
-*   Removed data from Queue.createJob().
-*   Added the Job.setPayload() method.
+*   Removed data from `Queue.createJob()`.
+*   Added the `Job.setPayload()` method.
 
@@ -76,0 +82,0 @@ ## v0.0.1 / 2016-07-27

dist/enums.js

@@ -18,3 +18,3 @@ 'use strict';
     medium: 30,
-    high: 20, // Used for retries after a job has timed out or failed.
+    high: 20,
     highest: 10
@@ -95,4 +95,5 @@ },
     dbError: 'RethinkDB returned an error',
-    concurrencyInvalid: 'Invalid concurrency value'
+    concurrencyInvalid: 'Invalid concurrency value',
+    cancelCallbackInvalid: 'The onCancel callback is not a function'
   }
 };

dist/queue-process.js

@@ -6,2 +6,3 @@ 'use strict';
 var enums = require('./enums');
+var is = require('./is');
 var dbReview = require('./db-review');
@@ -13,5 +14,6 @@ var queueGetNextJob = require('./queue-get-next-job');
 var jobTimeouts = new Map();
+var jobOnCancelHandlers = new Map();
 
 function addJobTimeout(job, timeoutHandler) {
-  logger('addJobTimeout');
+  logger('addJobTimeout', job);
   var timeoutValue = job.timeout;
@@ -26,4 +28,16 @@ var jobTimeout = {
 
-function removeJobTimeout(jobId) {
-  logger('removeJobTimeout');
+function addOnCancelHandler(job, cancellationCallback) {
+  logger('addJobCancellation', job.id);
+  if (is.function(cancellationCallback)) {
+    jobOnCancelHandlers.set(job.id, cancellationCallback);
+  } else {
+    var err = new Error(enums.message.cancelCallbackInvalid);
+    logger('Event: error [' + err + ']');
+    job.q.emit(enums.status.error, err);
+    throw err;
+  }
+}
+
+function removeJobTimeoutAndOnCancelHandler(jobId) {
+  logger('removeJobTimeoutAndOnCancelHandler', jobId);
   if (jobTimeouts.has(jobId)) {
@@ -34,6 +48,19 @@ var jobTimeout = jobTimeouts.get(jobId);
   }
+  jobOnCancelHandlers.delete(jobId);
 }
 
+function onCancelJob(jobId, q) {
+  logger('onCancelJob', jobId);
+  if (jobOnCancelHandlers.has(jobId)) {
+    var onCancelHandler = jobOnCancelHandlers.get(jobId);
+    removeJobTimeoutAndOnCancelHandler(jobId);
+    q._running--;
+    // Calling the user defined cancel function
+    onCancelHandler();
+    setImmediate(jobTick, q);
+  }
+}
+
 function restartJobTimeout(jobId) {
-  logger('resetJobTimeout');
+  logger('resetJobTimeout', jobId);
   var jobTimeout = void 0;
@@ -60,3 +87,3 @@ if (jobTimeouts.has(jobId)) {
 
-    removeJobTimeout(job.id);
+    removeJobTimeoutAndOnCancelHandler(job.id);
 
@@ -68,3 +95,3 @@ return new Promise(function (resolve, reject) {
       }
-      resolve(jobResult);
+      return resolve(jobResult);
     }).then(function (successResult) {
@@ -101,3 +128,3 @@ logger('jobResult resolved successfully');
   logger('calling handler function');
-  job.q._handler(job, nextHandler);
+  job.q._handler(job, nextHandler, addOnCancelHandler);
 }
@@ -171,2 +198,5 @@
   q.on(enums.status.progress, restartJobTimeout);
+  q.on(enums.status.cancelled, function (jobId) {
+    return onCancelJob(jobId, q);
+  });
 
@@ -173,0 +203,0 @@ return Promise.resolve().then(function () {

package.json

 {
   "name": "rethinkdb-job-queue",
-  "version": "0.1.2",
+  "version": "0.2.0",
   "description": "A persistent job or task queue backed by RethinkDB.",
@@ -5,0 +5,0 @@ "main": "index.js",

README.md

@@ -10,2 +10,3 @@ # Introduction
 [![js-standard-style][js-standard-image]][js-standard-url]
+[![NSP Status][nsp-image]][nsp-url]
 
@@ -27,14 +28,15 @@ [![Thinker][thinker-image]][rjq-github-url]
 *   Promise based with minimal callbacks
-*   Jobs support:
-   *   [Priority processing][job-priority-url]
-   *   [Progress updates][job-progress-url]
-   *   [Delayed start][job-delayed-url]
-   *   [Cancelling][job-cancel-url]
-   *   [Timeout][job-timeout-url]
-   *   [Retrying][job-retry-url]
-   *   [Reanimation][job-reanimation-url]
-   *   [Editing][job-editing-url]
-   *   Rich [history log][job-log-url]
-* Over 1100 [integration tests][testing-url]
+*   [Priority processing of jobs][job-priority-url]
+*   [Job progress updates][job-progress-url]
+*   [Delayed job start][job-delayed-url]
+*   [Global job cancellation][job-cancel-url]
+*   [Job timeout][job-timeout-url]
+*   [Retrying failed jobs][job-retry-url]
+*   [Job reanimation][job-reanimation-url]
+*   [Job Editing][job-editing-url]
+*   Rich job [history log][job-log-url]
+*   Over 1200 [integration tests][testing-url]
 
+
+
 [queue-connection-url]: https://github.com/grantcarthew/node-rethinkdb-job-queue/wiki/Queue-Connection
@@ -62,3 +64,3 @@ [queue-events-url]: https://github.com/grantcarthew/node-rethinkdb-job-queue/wiki/Queue-Events
 *   This `rethinkdb-job-queue` module is fully functional.
-*   There are over 1100 integration tests.
+*   There are over 1200 integration tests.
 *   This project is complete and needs to be taken out for a spin.
@@ -183,3 +185,3 @@ *   In a few months I will bump the version up to 1.0.0 to support [SemVer](http://semver.org/).
   }).catch((err) => {
-    // This catch if for nodemailer sendMail errors.
+    // This catch is for nodemailer sendMail errors.
     return next(err)
@@ -244,3 +246,5 @@ })
 [js-standard-url]: http://standardjs.com/
+[nsp-image]: https://nodesecurity.io/orgs/openjs/projects/3871d340-0ca9-471c-be9a-39df3871262d/badge
+[nsp-url]: https://nodesecurity.io/orgs/openjs/projects/3871d340-0ca9-471c-be9a-39df3871262d
 [nodei-npm-image]: https://nodei.co/npm/rethinkdb-job-queue.png?downloads=true&downloadRank=true&stars=true
 [nodei-npm-url]: https://nodei.co/npm/rethinkdb-job-queue/

tests/enums.spec.js

@@ -21,3 +21,3 @@ const test = require('tape')
       t.equal(Object.keys(enums.log).length, 3, 'Enums log has has correct number of keys')
-      t.equal(Object.keys(enums.message).length, 12, 'Enums message has has correct number of keys')
+      t.equal(Object.keys(enums.message).length, 13, 'Enums message has has correct number of keys')
     } catch (err) {
@@ -24,0 +24,0 @@ tError(err, module, t)

tests/queue-get-next-job.spec.js

@@ -48,3 +48,3 @@ const test = require('tape')
       jobFailed.data = 'Failed'
-      jobFailed.dateCreated = datetime.add.sec(new Date(), -1)
+      jobFailed.dateCreated = datetime.add.sec(new Date(), -10)
       const jobActive = q.createJob({priority: 'normal'})
@@ -51,0 +51,0 @@ jobActive.status = enums.status.active

tests/queue-process.spec.js

@@ -16,6 +16,7 @@ const test = require('tape')
     test('queue-process', (t) => {
-      t.plan(207)
+      t.plan(216)
 
       // ---------- Test Setup ----------
       const q = new Queue(tOpts.cxn(), tOpts.default())
+      let qGlobalCancel
 
@@ -53,3 +54,3 @@ let jobs
       let processingEventCount = 0
-      const processingEventTotal = 37
+      const processingEventTotal = 38
       function processingEventHandler (jobId) {
@@ -81,3 +82,3 @@ if (testEvents) {
       let cancelledEventCount = 0
-      const cancelledEventTotal = 1
+      const cancelledEventTotal = 2
       function cancelledEventHandler (jobId) {
@@ -91,3 +92,3 @@ if (testEvents) {
       let idleEventCount = 0
-      const idleEventTotal = 11
+      const idleEventTotal = 12
       function idleEventHandler (qid) {
@@ -145,3 +146,3 @@ if (testEvents) {
       const summaryCompleted = 32
-      const summaryCancelled = 1
+      const summaryCancelled = 2
       const summaryTerminated = 1
@@ -153,3 +154,5 @@
       let testCancel = false
-      function testHandler (job, next) {
+      let jobProcessTimeoutId = false
+
+      function testHandler (job, next, onCancel) {
         if (testTimes) {
@@ -161,2 +164,3 @@ const testDate = datetime.add.ms(new Date(),
         }
+
         t.pass(`Job Started: Delay: [${jobDelay}] ID: [${job.id}]`)
@@ -176,3 +180,4 @@ if (testCancel) {
 
-          setTimeout(function () {
+          jobProcessTimeoutId = setTimeout(function () {
+            jobProcessTimeoutId = false
             next('Job Completed: ' + job.id)
@@ -183,2 +188,9 @@ .then((runningJobs) => {
           }, jobDelay)
+
+          return onCancel(job, () => {
+            clearTimeout(jobProcessTimeoutId)
+            jobProcessTimeoutId = false
+            t.pass('onCancel invoked')
+            return
+          })
         }
@@ -301,2 +313,23 @@ }
 
+        // ---------- Processing with Global Cancel Test ----------
+        t.comment('queue-process: Processing with Global Cancel')
+        qGlobalCancel = new Queue(tOpts.cxn(), tOpts.default())
+        jobDelay = 10000
+        jobs = q.createJob()
+        return q.addJob(jobs)
+      }).delay(1000).then(() => {
+        return q.getJob(jobs)
+      }).then((globalCancelJobs1) => {
+        t.equal(globalCancelJobs1[0].status, enums.status.active, 'Job status is active')
+        t.ok(jobProcessTimeoutId, 'Job is being processed')
+        return qGlobalCancel.cancelJob(globalCancelJobs1[0])
+      }).delay(1000).then(() => {
+        return q.getJob(jobs)
+      }).then((globalCancelJobs2) => {
+        t.equal(globalCancelJobs2[0].status, enums.status.cancelled, 'Job status is cancelled')
+        t.notOk(jobProcessTimeoutId, 'Job processing has been stopped')
+        jobDelay = 200
+        return qGlobalCancel.stop()
+      }).then(() => {
+        //
         // ---------- Delayed Job Start Test ----------
@@ -303,0 +336,0 @@ t.comment('queue-process: Delayed Job Start')

tests/test-current.js

@@ -12,3 +12,3 @@ // const Promise = require('bluebird')
 // const dbReview = require('./db-review.spec')
-const job = require('./job.spec')
+// const job = require('./job.spec')
 // const jobOptions = require('./job-options.spec')
@@ -26,3 +26,3 @@ // const jobParse = require('./job-parse.spec')
 // const queueGetNextJob = require('./queue-get-next-job.spec')
-// const queueProcess = require('./queue-process.spec')
+const queueProcess = require('./queue-process.spec')
 // const queueChange = require('./queue-change.spec')
@@ -39,3 +39,3 @@ // const queueInterruption = require('./queue-interruption.spec')
 }).then(() => {
-  return job()
+  return queueProcess()
 })

No alert changes

Improved metrics

Total package byte prevSize

247810

increased by1.16%

Lines of code

5662

increased by1.05%

Number of lines in readme file

246

increased by1.65%

No dependency changes