refactored - less closure; added pool.poolState

Author: Eetay Natan

README.md

@@ -113,7 +113,6 @@ pool.then(function(result) {
 ```
 
 ## TODO list:
-- refactor to use the master promise itself as the state holder rather than a separate closure
 - API similar to Promise.all(): "Promise.pool([promise, ...], threads)"
 - options for behavior on reject:
   - reject as soon as one is rejected (same as Promise.all)

promise-pool.js

@@ -1,51 +1,52 @@
-function promisePool(options) {
-  function PromisePoolResolver({max_parallel, next_promise, next_promise_data, threads}) {
-    const self = this
-    this.resolver = function(resolve, _reject) {
-      self.max = max_parallel || threads
-      self.started = 0
-      self.ended = 0
-      self.next_promise = Array.isArray(next_promise) ? [...next_promise] : next_promise
-      self.next_promise_data = next_promise_data
-      self.results = []
-      function startNext(self, thread) {
-        const context = {
-          index: self.started,
-          thread,
-          data: self.next_promise_data,
-          ended: false
-        }
-        const next = Array.isArray(self.next_promise) ? self.next_promise.shift() : self.next_promise({ index: self.started, data: self.next_promise_data })
-        self.started += 1
-        if (next && next.then) {
-          //console.log('promise ' + JSON.stringify(context))
-          next.then(function(result) {
-            context.ended = self.ended
-            self.ended += 1
-            //console.log(`promise ${context.index} resolved`)
-            self.results[context.index] = { context, promise: next, result: result }
-            startNext(self, thread)
-          }).catch(function(err) {
-            context.ended = self.ended
-            self.ended += 1
-            //console.log(`promise ${context.index} rejected`)
-            self.results[context.index] = { context, promise: next, error: err }
-            startNext(self, thread)
-          })
-        } else {
-          self.live -= 1
-          if (self.live <= 0) {
-            resolve(self.results)
-          }
-        }
+function promisePool({max_parallel, next_promise, next_promise_data, threads, promises, user_data}) {
+  var promises_generator = promises || next_promise
+  var self = {
+    threads: max_parallel || threads,
+    started: 0,
+    ended: 0,
+    promises_generator: Array.isArray(promises_generator) ? [...promises_generator] : promises_generator,
+    next_promise_data: next_promise_data || user_data,
+    results: []
+  }
+  self.live = self.threads
+  var promise = new Promise(function(resolve, _reject) {
+    function startNext(self, thread) {
+      const context = {
+        index: self.started,
+        thread,
+        data: self.next_promise_data,
+        ended: false
       }
-      self.live = self.max
-      for (var i=0; i<self.max; i+=1) {
-        startNext(self, i)
+      const next = Array.isArray(self.promises_generator) ? self.promises_generator.shift() : self.promises_generator({ index: self.started, data: self.next_promise_data })
+      self.started += 1
+      if (next && next.then) {
+        //console.log('promise ' + JSON.stringify(context))
+        next.then(function(result) {
+          context.ended = self.ended
+          self.ended += 1
+          //console.log(`promise ${context.index} resolved`)
+          self.results[context.index] = { context, promise: next, result: result }
+          startNext(self, thread)
+        }).catch(function(err) {
+          context.ended = self.ended
+          self.ended += 1
+          //console.log(`promise ${context.index} rejected`)
+          self.results[context.index] = { context, promise: next, error: err }
+          startNext(self, thread)
+        })
+      } else {
+        self.live -= 1
+        if (self.live <= 0) {
+          resolve(self.results)
+        }
       }
     }
-  }
-  return new Promise(new PromisePoolResolver(options).resolver)
+    for (var i=0; i<self.threads; i+=1) {
+      startNext(self, i)
+    }
+  })
+  promise.poolState = self
+  return promise
 }
 
 module.exports = promisePool

spec/promise-pool.spec.js

@@ -6,19 +6,16 @@ function randomTimeout() {
 }
 
 function makePromise(i, timeout = 0) {
-  const closure = function () {
-    var context = {}
-    var promise = new Promise(function (resolve, _reject) {
-      setTimeout(() => {
-        context.promise.done = true
-        resolve(i)
-      }, timeout)
-    })
-    context.promise = promise
-    context.promise.done = false
-    return promise
-  }
-  return closure()
+  var context = {}
+  var promise = new Promise(function (resolve, _reject) {
+    setTimeout(() => {
+      context.promise.done = true
+      resolve(i)
+    }, timeout)
+  })
+  context.promise = promise
+  promise.done = false
+  return promise
 }
 
 test('Nested promise pools', (done) => {
@@ -45,7 +42,6 @@ test('Nested promise pools', (done) => {
     next_promise_data: 'primary promise pool'
   })
   pool.then(function(result) {
-    console.log(JSON.stringify(result))
     expect(result.length).toBe(primaryPromiseList.length)
     const secondaryResult = result[2].result
     expect(secondaryResult.length).toBe(secondaryPromiseList.length)
@@ -75,7 +71,7 @@ test('Array of promises', (done) => {
   })
 })
 
-test('Array of promises; one is not resolved', (done) => {
+test('Array of promises; one is not done yet', (done) => {
   expect.assertions(6)
   var resolved=false
   const promiseList = [
@@ -105,7 +101,7 @@ test('Array of promises; one is not resolved', (done) => {
   })
 })
 
-test('20 promises; max parallel 3', (done) => {
+test('20 promises; threads parallel 3', (done) => {
   expect.assertions(2)
   const numPromises = 20
   const pool = promisePool({