Simplify listener subscription logic

Always remove pending listeners on acquire call.
Always register add and release listeners on acquire.

Author: rochoa

batch/scheduler/scheduler.js

@@ -109,6 +109,9 @@ Scheduler.prototype.schedule = function() {
 };
 
 Scheduler.prototype.acquire = function(callback) {
+    this.removeAllListeners('add');
+    this.removeAllListeners('release');
+
     if (this.tasks.every(is(STATUS.DONE))) {
         return callback(null, null);
     }
@@ -119,39 +122,29 @@ Scheduler.prototype.acquire = function(callback) {
             return callback(err);
         }
 
-        function addListener() {
-            self.removeListener('release', releaseListener);
+        debug('Trying to acquire task');
+        var running = self.tasks.filter(is(STATUS.RUNNING));
+        debug('[capacity=%d, running=%d] candidates=%j', capacity, running.length, self.tasks);
+
+        self.once('add', function() {
             debug('Got a new task');
             self.acquire(callback);
-        }
-
-        function releaseListener() {
-            self.removeListener('add', addListener);
+        });
+        self.once('release', function() {
             debug('Slot was released');
             self.acquire(callback);
-        }
-
-        debug('Trying to acquire task');
-
-        var allRunning = self.tasks.every(is(STATUS.RUNNING));
-        var running = self.tasks.filter(is(STATUS.RUNNING));
-        debug('[capacity=%d, running=%d, all=%s] candidates=%j', capacity, running.length, allRunning, self.tasks);
-
-        if (allRunning && running.length < capacity) {
-            debug('Waiting for tasks');
-            self.once('add', addListener);
-        }
+        });
 
         if (running.length >= capacity) {
-            debug('Waiting for slot');
-            return self.once('release', releaseListener);
+            debug('Not enough capacity');
+            return null;
         }
 
         var isRunningAny = self.tasks.some(is(STATUS.RUNNING));
         var candidate = self.tasksTree.min();
         if (isRunningAny && candidate === null) {
             debug('Waiting for last task to finish');
-            return self.once('release', releaseListener);
+            return null;
         }
 
         return callback(null, candidate);

test/integration/batch/scheduler.js

@@ -21,11 +21,11 @@ describe('scheduler', function() {
     };
 
     // simulate one by one or infinity capacity
-    var capacities = [new FixedCapacity(1), new FixedCapacity(Infinity)];
+    var capacities = [new FixedCapacity(1), new FixedCapacity(2), new FixedCapacity(Infinity)];
 
     capacities.forEach(function(capacity) {
 
-        it('regression', function (done) {
+        it('regression #1', function (done) {
             var taskRunner = new TaskRunner({
                 userA: 2,
                 userB: 2
@@ -50,6 +50,44 @@ describe('scheduler', function() {
             scheduler.schedule();
         });
 
+        it('regression #2', function (done) {
+            var taskRunner = new TaskRunner({
+                userA: 2,
+                userB: 2,
+                userC: 2,
+                userD: 1
+            });
+            var scheduler = new Scheduler(capacity, taskRunner);
+            scheduler.add('userA');
+            scheduler.add('userB');
+
+            scheduler.on('done', function() {
+                var results = taskRunner.results;
+
+                assert.equal(results.length, 7);
+
+                assert.equal(results[0], 'userA');
+                assert.equal(results[1], 'userB');
+                assert.equal(results[2], 'userC');
+                assert.equal(results[3], 'userD');
+                assert.equal(results[4], 'userA');
+                assert.equal(results[5], 'userB');
+                assert.equal(results[6], 'userC');
+
+                return done();
+            });
+
+            setTimeout(function() {
+                scheduler.add('userC');
+            }, 10);
+
+            setTimeout(function() {
+                scheduler.add('userD');
+            }, 20);
+
+            scheduler.schedule();
+        });
+
         it('should run tasks', function (done) {
             var taskRunner = new TaskRunner({
                 userA: 1