API change - require consumers to be returned to the scheduler so tha…

…t they are not spun down whilst in use.

README.md

@@ -34,6 +34,7 @@ var scheduler = new RoundRobin({
 var intervalId = setInterval(function () {
   scheduler.get(function (er, thing) {
     thing.doTask();
+    scheduler.ret(thing); // Return the thing so the scheduler can hand it out again
   })
 }, 1000);
 

lib/round-robin.js

@@ -48,7 +48,7 @@ util.inherits(RoundRobin, events.EventEmitter)
 // Callback gets a boolean to say whether the consumer was spun down or not
 function spinDownIf (consumer, cb) {
   // Do the spin down if expired or done too much work
-  if (consumer.expires < Date.now() || consumer.usage >= this.opts.maxUsage) {
+  if (!consumer.inUse && (consumer.expires < Date.now() || consumer.usage >= this.opts.maxUsage)) {
     this.consumers.splice(this.consumers.indexOf(consumer), 1)
 
     return this.opts.spinDown(consumer.consumer, function (er) {
@@ -93,6 +93,8 @@ RoundRobin.prototype.get = function (cb) {
 
     if (consumer) {
       consumer.usage++
+      consumer.inUse = true
+
       return cb(null, consumer.consumer)
     }
 
@@ -105,14 +107,22 @@ RoundRobin.prototype.get = function (cb) {
       }.bind(this))
     }
 
+    // Spinning down might have caused the current consumer pointer to point outside the array bounds
+    if (this.currentConsumer >= this.consumers.length) {
+      this.currentConsumer = 0
+    }
+
     consumer = this.consumers[this.currentConsumer]
 
     //console.log("Might return consumer", this.currentConsumer, "of", this.consumers.length)
 
     this.currentConsumer++
-
-    if (this.currentConsumer >= this.consumers.length) {
-      this.currentConsumer = 0
+
+    if (consumer.inUse) {
+      // Once an in use consumer has been returned, use it!
+      return this.once("ret", function () {
+        this.get(cb)
+      }.bind(this))
     }
 
     // If we're lazily spinning down, we need to check if this consumer has expired or is used up
@@ -121,25 +131,35 @@ RoundRobin.prototype.get = function (cb) {
         if (er) return cb(er)
 
         if (didSpinDown) {
-          // Spinning down might have caused the current consumer pointer to point outside the array bounds
-          if (this.currentConsumer >= this.consumers.length) {
-            this.currentConsumer = 0
-          }
           return this.get(cb)
         }
 
         consumer.usage++
+        consumer.inUse = true
+
         cb(null, consumer.consumer)
 
       }.bind(this))
     }
 
     consumer.usage++
+    consumer.inUse = true
+
     cb(null, consumer.consumer)
 
   }.bind(this))
 }
 
+RoundRobin.prototype.ret = function (consumer) {
+  for (var i = 0; i < this.consumers.length; ++i) {
+    if (this.consumers[i].consumer === consumer) {
+      this.consumers[i].inUse = false
+      return this.emit("ret")
+    }
+  }
+  throw new Error("Tried to return a consumer the scheduler didn't own")
+}
+
 RoundRobin.prototype.destroy = function (cb) {
 
   clearInterval(this.eagerSpinDownIntervalId)
@@ -157,6 +177,7 @@ function Consumer (consumer, maxAge) {
   this.consumer = consumer
   this.expires = Date.now() + maxAge
   this.usage = 0
+  this.inUse = false
 }
 
 module.exports.RoundRobin = RoundRobin