keuss - npm Package Compare versions

Comparing version 1.0.1 to 1.0.2

AsyncQueue.js

@@ -308,12 +308,26 @@ 'use strict';
   /////////////////////////////
-  _nextDelta () {
+  _nextDelta (cb) {
   /////////////////////////////
     if (!this._next_mature_t) {
-      this._verbose  ('_nextDelta: default delta is %d msecs', this._pollInterval);
-      return this._pollInterval;
+      var self = this;
+
+      this.next_t (function (err, res) {
+        if (err) return cb (this._pollInterval);
+
+        if (res) {
+          var delta = res - Queue.now ().getTime();
+          self._verbose  ('_nextDelta: got from queue.next_t to be %d msecs', delta);
+          return cb (delta);
+        }
+        else {
+          self._verbose  ('_nextDelta: null queue.next_t, use default poll interval (%d msecs)', self._pollInterval);
+          return cb (self._pollInterval);
+        }
+      });
     }
-      
-    var delta = this._next_mature_t - Queue.now ().getTime();
-    this._verbose  ('_nextDelta: explicit delta is %d msecs', delta);
-    return delta;
+    else {
+      var delta = this._next_mature_t - Queue.now ().getTime();
+      this._verbose  ('_nextDelta: explicit delta is %d msecs', delta);
+      return cb (delta);
+    }
   }
@@ -326,3 +340,3 @@
   //////////////////////////////////
-    // queue is empty: push consumer again
+    // queue is empty or non-mature: push consumer again
     if (consumer) {
@@ -341,9 +355,9 @@ this._consumers_by_order.push (consumer);
     }
-    
-    var delta = this._nextDelta ();
-
-    //|| TODO cap out delta to a max of 5 or 10 min
+     
     var self = this;
-    this._getOrFail_timeout = setTimeout (function () {self._getOrFail ()}, delta);
-    this._verbose  ('_rearm_getOrFail: _getOrFail rearmed, wait is %d', delta);
+    this._nextDelta (function (delta) {
+      //|| TODO cap out delta to a max of 5 or 10 min
+      self._getOrFail_timeout = setTimeout (function () {self._getOrFail ()}, delta);
+      self._verbose  ('_rearm_getOrFail: _getOrFail rearmed, wait is %d', delta);
+    });
   }
@@ -428,2 +442,3 @@
         return self._reinsertAndRearm (result);
+        // TODO do not reinsert if it's using reserve!!!!
       }
@@ -430,0 +445,0 @@

backends/mongo.js

@@ -62,3 +62,3 @@ 'use strict';
     var self = this;
-    this._col.findOneAndDelete ({}, {sort: {mature : 1}}, function (err, result) {
+    this._col.findOneAndDelete ({mature: {$lte: Queue.nowPlusSecs (0)}}, {sort: {mature : 1}}, function (err, result) {
       if (err) {
@@ -81,2 +81,6 @@ return callback (err);
     
+    var query = {
+      mature: {$lte: Queue.nowPlusSecs (0)}
+    };
+
     var update = {
@@ -92,3 +96,3 @@ $set: {mature: Queue.nowPlusSecs (delay), reserved: new Date ()},
     
-    this._col.findOneAndUpdate ({}, update, opts, function (err, result) {
+    this._col.findOneAndUpdate (query, update, opts, function (err, result) {
       if (err) {
@@ -95,0 +99,0 @@ return callback (err);

package.json

 {
   "name": "keuss",
-  "version": "1.0.1",
+  "version": "1.0.2",
   "keywords": ["queue", "job"],
@@ -20,9 +20,9 @@ "homepage":"https://github.com/pepmartinez/keuss",
   "dependencies": {
-    "async": "2.4.1",
-    "mongodb": "2.2.28",
+    "async": "2.5.0",
+    "mongodb": "2.2.29",
     "lodash": "4.17.4",
     "winston": "2.3.1",
-    "uuid": "3.0.1",
+    "uuid": "3.1.0",
     "ioredis": "3.1.1",
-    "commander": "2.9.0"
+    "commander": "2.10.0"
   },
@@ -29,0 +29,0 @@ "devDependencies": {

signal/local.js

@@ -33,2 +33,4 @@ 'use strict';
 
+// TODO: make local, in-process pubsub over the factory. use mitt
+
 module.exports = LocalSignalFactory;

signal/redis-pubsub.js

@@ -50,2 +50,4 @@ 'use strict';
 
+// TODO: relay to in-process pubsub over the factory. use mitt
+
 module.exports = RPSSignalFactory;

test/backends_mongo.js

@@ -72,3 +72,3 @@
       function (cb) {q.next_t (function (err, res) {
-        res.getTime().should.be.approximately (new Date().getTime (), 100);
+        res.getTime().should.be.approximately (new Date().getTime (), 500);
         cb();
@@ -135,3 +135,3 @@ })},
       function (cb) {q.next_t (function (err, res) {
-        res.getTime().should.be.approximately (new Date().getTime () + 1000, 100);
+        res.getTime().should.be.approximately (new Date().getTime () + 1000, 500);
         cb();
@@ -138,0 +138,0 @@ })},

utils/RedisOrderedQueue.js

@@ -18,2 +18,3 @@ 'use strict';
   -- qname in KEYS[1]
+  -- mature_mark in ARGV[1]
   
@@ -28,3 +29,8 @@ -- get older (lower mature) id from index
   local id = z_res[1]
---  local mature = z_res[2]
+  local mature = z_res[2]
+
+  if (mature > ARGV[1]) then
+    -- head is not mature, just end
+    return nil
+  end
   
@@ -59,3 +65,3 @@ -- get val by id from hash
   pop (done) {
-    this._rediscl.roq_pop (this._name, function (err, res) {
+    this._rediscl.roq_pop (this._name, new Date().getTime (), function (err, res) {
       if (err) return done (err);
@@ -80,4 +86,3 @@
   //////////////////////////////////
-    var now = new Date();
-    this._rediscl.zcount ('keuss:q:ordered_queue:index:' + this._name, '-inf', now.getTime(), callback);
+    this._rediscl.zcount ('keuss:q:ordered_queue:index:' + this._name, '-inf', new Date().getTime(), callback);
   }
@@ -90,4 +95,3 @@
   //////////////////////////////////
-    var now = new Date();
-    this._rediscl.zcount ('keuss:q:ordered_queue:index:' + this._name, now.getTime(), '+inf', callback);
+    this._rediscl.zcount ('keuss:q:ordered_queue:index:' + this._name, new Date().getTime(), '+inf', callback);
   }
@@ -94,0 +98,0 @@

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

150242

increased by0.56%

Lines of code

3025

increased by0.77%

Dependency changes

• + Addedasync@2.5.0(transitive)

• + Addedcommander@2.10.0(transitive)

• + Addedmongodb@2.2.29(transitive)

• + Addedmongodb-core@2.1.13(transitive)

• + Addeduuid@3.1.0(transitive)

• - Removedasync@2.4.1(transitive)

• - Removedcommander@2.9.0(transitive)

• - Removedmongodb@2.2.28(transitive)

• - Removedmongodb-core@2.1.12(transitive)

• - Removeduuid@3.0.1(transitive)

• Updatedasync@2.5.0

• Updatedcommander@2.10.0

• Updatedmongodb@2.2.29

• Updateduuid@3.1.0