Fixed purging of all connections in a pool

And make sure connections, returned to the pool, are destroyed after
corresponding key has been purged.

Author: lutovich

src/v1/internal/pool.js

@@ -70,11 +70,7 @@ class Pool {
   }
 
   purgeAll() {
-    for (let key in this._pools.keys) {
-      if (this._pools.hasOwnPropertykey) {
-        this.purge(key);
-      }
-    }
+    Object.keys(this._pools).forEach(key => this.purge(key));
   }
 
   has(key) {
@@ -84,7 +80,8 @@ class Pool {
   _release(key, resource) {
     let pool = this._pools[key];
     if (!pool) {
-      //key has been purged, don't put it back
+      // key has been purged, don't put it back, just destroy the resource
+      this._destroy(resource);
       return;
     }
     if( pool.length >= this._maxIdle || !this._validate(resource) ) {
@@ -96,4 +93,3 @@ class Pool {
 }
 
 export default Pool
-

test/internal/pool.test.js

@@ -32,8 +32,9 @@ describe('Pool', () => {
     const r1 = pool.acquire(key);
 
     // Then
-    expect( r0.id ).toBe( 0 );
-    expect( r1.id ).toBe( 1 );
+    expect(r0.id).toBe(0);
+    expect(r1.id).toBe(1);
+    expect(r0).not.toBe(r1);
   });
 
   it('pools if resources are returned', () => {
@@ -48,8 +49,9 @@ describe('Pool', () => {
     const r1 = pool.acquire(key);
 
     // Then
-    expect( r0.id ).toBe( 0 );
-    expect( r1.id ).toBe( 0 );
+    expect(r0.id).toBe(0);
+    expect(r1.id).toBe(0);
+    expect(r0).toBe(r1);
   });
 
   it('handles multiple keys', () => {
@@ -67,10 +69,13 @@ describe('Pool', () => {
     const r3 = pool.acquire(key2);
 
     // Then
-    expect( r0.id ).toBe( 0 );
-    expect( r1.id ).toBe( 1 );
-    expect( r2.id ).toBe( 0 );
-    expect( r3.id ).toBe( 2 );
+    expect(r0.id).toBe(0);
+    expect(r1.id).toBe(1);
+    expect(r2.id).toBe(0);
+    expect(r3.id).toBe(2);
+
+    expect(r0).toBe(r2);
+    expect(r1).not.toBe(r3);
   });
 
   it('frees if pool reaches max size', () => {
@@ -96,8 +101,8 @@ describe('Pool', () => {
     r2.close();
 
     // Then
-    expect( destroyed.length ).toBe( 1 );
-    expect( destroyed[0].id ).toBe( r2.id );
+    expect(destroyed.length).toBe(1);
+    expect(destroyed[0].id).toBe(r2.id);
   });
 
   it('frees if validate returns false', () => {
@@ -121,9 +126,9 @@ describe('Pool', () => {
     r1.close();
 
     // Then
-    expect( destroyed.length ).toBe( 2 );
-    expect( destroyed[0].id ).toBe( r0.id );
-    expect( destroyed[1].id ).toBe( r1.id );
+    expect(destroyed.length).toBe(2);
+    expect(destroyed[0].id).toBe(r0.id);
+    expect(destroyed[1].id).toBe(r1.id);
   });
 
 
@@ -144,21 +149,111 @@ describe('Pool', () => {
     const r1 = pool.acquire(key2);
     r0.close();
     r1.close();
-    expect(pool.has(key1)).toBe(true);
-    expect(pool.has(key2)).toBe(true);
+    expect(pool.has(key1)).toBeTruthy();
+    expect(pool.has(key2)).toBeTruthy();
     pool.purge(key1);
-    expect(pool.has(key1)).toBe(false);
-    expect(pool.has(key2)).toBe(true);
+    expect(pool.has(key1)).toBeFalsy();
+    expect(pool.has(key2)).toBeTruthy();
 
     const r2 = pool.acquire(key1);
     const r3 = pool.acquire(key2);
 
     // Then
-    expect( r0.id ).toBe( 0 );
-    expect( r0.destroyed ).toBe( true );
-    expect( r1.id ).toBe( 1 );
-    expect( r2.id ).toBe( 2 );
-    expect( r3.id ).toBe( 1 );
+    expect(r0.id).toBe(0);
+    expect(r0.destroyed).toBeTruthy();
+    expect(r1.id).toBe(1);
+    expect(r2.id).toBe(2);
+    expect(r3.id).toBe(1);
+  });
+
+  it('destroys resource when key was purged', () => {
+    let counter = 0;
+    const key = 'bolt://localhost:7687';
+    const pool = new Pool((url, release) => new Resource(url, counter++, release),
+      res => {
+        res.destroyed = true;
+        return true;
+      }
+    );
+
+    const r0 = pool.acquire(key);
+    expect(pool.has(key)).toBeTruthy();
+    expect(r0.id).toEqual(0);
+
+    pool.purge(key);
+    expect(pool.has(key)).toBeFalsy();
+    expect(r0.destroyed).toBeFalsy();
+
+    r0.close();
+    expect(pool.has(key)).toBeFalsy();
+    expect(r0.destroyed).toBeTruthy();
+  });
+
+  it('purges all keys', () => {
+    let counter = 0;
+
+    const key1 = 'bolt://localhost:7687';
+    const key2 = 'bolt://localhost:7688';
+    const key3 = 'bolt://localhost:7689';
+
+    const pool = new Pool((url, release) => new Resource(url, counter++, release),
+      res => {
+        res.destroyed = true;
+        return true;
+      }
+    );
+
+    const acquiredResources = [
+      pool.acquire(key1),
+      pool.acquire(key2),
+      pool.acquire(key3),
+      pool.acquire(key1),
+      pool.acquire(key2),
+      pool.acquire(key3)
+    ];
+    acquiredResources.forEach(resource => resource.close());
+
+    pool.purgeAll();
+
+    acquiredResources.forEach(resource => expect(resource.destroyed).toBeTruthy());
+  });
+
+  it('skips broken connections during acquire', () => {
+    let validated = false;
+    let counter = 0;
+    const key = 'bolt://localhost:7687';
+    const pool = new Pool((url, release) => new Resource(url, counter++, release),
+      res => {
+        res.destroyed = true;
+        return true;
+      },
+      () => {
+        if (validated) {
+          return false;
+        }
+        validated = true;
+        return true;
+      }
+    );
+
+    const r0 = pool.acquire(key);
+    r0.close();
+
+    const r1 = pool.acquire(key);
+    expect(r1).not.toBe(r0);
+  });
+
+  it('reports presence of the key', () => {
+    const existingKey = 'bolt://localhost:7687';
+    const absentKey = 'bolt://localhost:7688';
+
+    const pool = new Pool((url, release) => new Resource(url, 42, release));
+
+    pool.acquire(existingKey);
+    pool.acquire(existingKey);
+
+    expect(pool.has(existingKey)).toBeTruthy();
+    expect(pool.has(absentKey)).toBeFalsy();
   });
 });
 
@@ -168,9 +263,10 @@ class Resource {
     this.id = id;
     this.key = key;
     this.release = release;
+    this.destroyed = false;
   }
 
   close() {
-    this.release(key, self);
+    this.release(this.key, this);
   }
 }

test/v1/routing.driver.boltkit.it.js

@@ -747,13 +747,19 @@ describe('routing driver', () => {
       const session1 = driver.session(neo4j.session.WRITE);
       session1.run("CREATE (n {name:'Bob'})").then(() => {
         session1.close(() => {
-          const connections = Object.keys(driver._openSessions).length;
+          const openConnectionsCount = numberOfOpenConnections(driver);
           const session2 = driver.session(neo4j.session.WRITE);
           session2.run("CREATE ()").then(() => {
+            // driver should have same amount of open connections at this point;
+            // no new connections should be created, existing connections should be reused
+            expect(numberOfOpenConnections(driver)).toEqual(openConnectionsCount);
             driver.close();
+
+            // all connections should be closed when driver is closed
+            expect(numberOfOpenConnections(driver)).toEqual(0);
+
             seedServer.exit(code1 => {
               writeServer.exit(code2 => {
-                expect(connections).toEqual(Object.keys(driver._openSessions).length);
                 expect(code1).toEqual(0);
                 expect(code2).toEqual(0);
                 done();
@@ -2064,6 +2070,10 @@ describe('routing driver', () => {
     return '[' + array.map(s => '"' + s + '"').join(',') + ']';
   }
 
+  function numberOfOpenConnections(driver) {
+    return Object.keys(driver._openSessions).length;
+  }
+
   class MemorizingRoutingTable extends RoutingTable {
 
     constructor(initialTable) {