Refactory Pool module (#930)

The state of the pool and the resource list weren't bounded together as it supposed to. This scenario makes the code prune to bugs and inconsistency in the long run.

Merge the resource list (a.k.a. Pool) and PoolState in the same object makes the code easier to understand and more cohesive.

Author: bigmontz

packages/bolt-connection/src/pool/pool.js

@@ -62,7 +62,6 @@ class Pool {
     this._pendingCreates = {}
     this._acquireRequests = {}
     this._activeResourceCounts = {}
-    this._poolState = {}
     this._release = this._release.bind(this)
     this._log = log
     this._closed = false
@@ -111,7 +110,6 @@ class Pool {
 
       request = new PendingRequest(key, resolve, reject, timeoutId, this._log)
       allRequests[key].push(request)
-
       this._processPendingAcquireRequests(address)
     })
   }
@@ -177,21 +175,23 @@ class Pool {
     return this._activeResourceCounts[address.asKey()] || 0
   }
 
+  _getOrInitializePoolFor (key) {
+    let pool = this._pools[key]
+    if (!pool) {
+      pool = new SingleAddressPool()
+      this._pools[key] = pool
+      this._pendingCreates[key] = 0
+    }
+    return pool
+  }
+
   async _acquire (address) {
     if (this._closed) {
       throw newError('Pool is closed, it is no more able to serve requests.')
     }
 
     const key = address.asKey()
-    let pool = this._pools[key]
-    let poolState = this._poolState[key]
-    if (!pool) {
-      pool = []
-      poolState = new PoolState()
-      this._pools[key] = pool
-      this._pendingCreates[key] = 0
-      this._poolState[key] = poolState
-    }
+    const pool = this._getOrInitializePoolFor(key)
     while (pool.length) {
       const resource = pool.pop()
 
@@ -205,7 +205,7 @@ class Pool {
         if (this._log.isDebugEnabled()) {
           this._log.debug(`${resource} acquired from the pool ${key}`)
         }
-        return resource
+        return { resource, pool }
       } else {
         await this._destroy(resource)
       }
@@ -219,7 +219,7 @@ class Pool {
         this.activeResourceCount(address) + this._pendingCreates[key]
       if (numConnections >= this._maxSize) {
         // Will put this request in queue instead since the pool is full
-        return null
+        return { resource: null, pool }
       }
     }
 
@@ -229,7 +229,7 @@ class Pool {
     let resource
     try {
       // Invoke callback that creates actual connection
-      resource = await this._create(address, (address, resource) => this._release(poolState, address, resource))
+      resource = await this._create(address, (address, resource) => this._release(address, resource, pool))
 
       resourceAcquired(key, this._activeResourceCounts)
       if (this._log.isDebugEnabled()) {
@@ -238,14 +238,13 @@ class Pool {
     } finally {
       this._pendingCreates[key] = this._pendingCreates[key] - 1
     }
-    return resource
+    return { resource, pool }
   }
 
-  async _release (poolState, address, resource) {
+  async _release (address, resource, pool) {
     const key = address.asKey()
-    const pool = this._pools[key]
 
-    if (pool && poolState.isActive()) {
+    if (pool.isActive()) {
       // there exist idle connections for the given key
       if (!this._validate(resource)) {
         if (this._log.isDebugEnabled()) {
@@ -292,24 +291,23 @@ class Pool {
   }
 
   async _purgeKey (key) {
-    const pool = this._pools[key] || []
-    const poolState = this._poolState[key] || new PoolState()
-    while (pool.length) {
-      const resource = pool.pop()
-      if (this._removeIdleObserver) {
-        this._removeIdleObserver(resource)
+    const pool = this._pools[key]
+    if (pool) {
+      while (pool.length) {
+        const resource = pool.pop()
+        if (this._removeIdleObserver) {
+          this._removeIdleObserver(resource)
+        }
+        await this._destroy(resource)
       }
-      await this._destroy(resource)
+      pool.close()
+      delete this._pools[key]
     }
-    poolState.close()
-    delete this._pools[key]
-    delete this._poolState[key]
   }
 
   _processPendingAcquireRequests (address) {
     const key = address.asKey()
     const requests = this._acquireRequests[key]
-    const poolState = this._poolState[key] || new PoolState()
     if (requests) {
       const pendingRequest = requests.shift() // pop a pending acquire request
 
@@ -319,16 +317,16 @@ class Pool {
             // failed to acquire/create a new connection to resolve the pending acquire request
             // propagate the error by failing the pending request
             pendingRequest.reject(error)
-            return null
+            return { resource: null }
           })
-          .then(resource => {
+          .then(({ resource, pool }) => {
             if (resource) {
               // managed to acquire a valid resource from the pool
 
               if (pendingRequest.isCompleted()) {
                 // request has been completed, most likely failed by a timeout
                 // return the acquired resource back to the pool
-                this._release(poolState, address, resource)
+                this._release(address, resource, pool)
               } else {
                 // request is still pending and can be resolved with the newly acquired resource
                 pendingRequest.resolve(resource) // resolve the pending request with the acquired resource
@@ -415,9 +413,10 @@ class PendingRequest {
   }
 }
 
-class PoolState {
+class SingleAddressPool {
   constructor () {
     this._active = true
+    this._elements = []
   }
 
   isActive () {
@@ -427,6 +426,23 @@ class PoolState {
   close () {
     this._active = false
   }
+
+  filter (predicate) {
+    this._elements = this._elements.filter(predicate)
+    return this
+  }
+
+  get length () {
+    return this._elements.length
+  }
+
+  pop () {
+    return this._elements.pop()
+  }
+
+  push (element) {
+    return this._elements.push(element)
+  }
 }
 
 export default Pool