Simplify connection release when closing session

This commit makes `Session#close()` responsible for releasing the underlying
connection to the pool. Previously this behaviour was passed in as a function
from the driver.

Author: lutovich

src/v1/driver.js

@@ -17,11 +17,11 @@
  * limitations under the License.
  */
 
-import Session from "./session";
-import Pool from "./internal/pool";
-import {connect} from "./internal/connector";
-import StreamObserver from "./internal/stream-observer";
-import {newError, SERVICE_UNAVAILABLE} from "./error";
+import Session from './session';
+import Pool from './internal/pool';
+import {connect} from './internal/connector';
+import StreamObserver from './internal/stream-observer';
+import {newError, SERVICE_UNAVAILABLE} from './error';
 
 const READ = 'READ', WRITE = 'WRITE';
 /**
@@ -119,33 +119,7 @@ class Driver {
         //we don't need to tell the driver about this error
       }
     });
-    return this._createSession(connectionPromise, this._releaseConnection(connectionPromise));
-  }
-
-  /**
-   * The returned function gets called on Session#close(), and is where we return the pooled 'connection' instance.
-   * We don't pool Session instances, to avoid users using the Session after they've called close.
-   * The `Session` object is just a thin wrapper around Connection anyway, so it makes little difference.
-   * @param {Promise} connectionPromise - promise resolved with the connection.
-   * @return {function(callback: function)} - function that releases the connection and then executes an optional callback.
-   * @protected
-   */
-  _releaseConnection(connectionPromise) {
-    return userDefinedCallback => {
-      connectionPromise.then(conn => {
-        // Queue up a 'reset', to ensure the next user gets a clean session to work with.
-        conn.reset();
-        conn.sync();
-
-        // Return connection to the pool
-        conn._release();
-      }).catch(ignoredError => {
-      });
-
-      if (userDefinedCallback) {
-        userDefinedCallback();
-      }
-    };
+    return this._createSession(connectionPromise);
   }
 
   static _validateSessionMode(rawMode) {
@@ -162,8 +136,8 @@ class Driver {
   }
 
   //Extension point
-  _createSession(connectionPromise, cb) {
-    return new Session(connectionPromise, cb);
+  _createSession(connectionPromise) {
+    return new Session(connectionPromise);
   }
 
   /**

src/v1/routing-driver.js

@@ -17,12 +17,12 @@
  * limitations under the License.
  */
 
-import Session from "./session";
-import {Driver, READ, WRITE} from "./driver";
-import {newError, SERVICE_UNAVAILABLE, SESSION_EXPIRED} from "./error";
-import RoundRobinArray from "./internal/round-robin-array";
-import RoutingTable from "./internal/routing-table";
-import Rediscovery from "./internal/rediscovery";
+import Session from './session';
+import {Driver, READ, WRITE} from './driver';
+import {newError, SERVICE_UNAVAILABLE, SESSION_EXPIRED} from './error';
+import RoundRobinArray from './internal/round-robin-array';
+import RoutingTable from './internal/routing-table';
+import Rediscovery from './internal/rediscovery';
 
 /**
  * A driver that supports routing in a core-edge cluster.
@@ -35,8 +35,8 @@ class RoutingDriver extends Driver {
     this._rediscovery = new Rediscovery();
   }
 
-  _createSession(connectionPromise, cb) {
-    return new RoutingSession(connectionPromise, cb, (error, conn) => {
+  _createSession(connectionPromise) {
+    return new RoutingSession(connectionPromise, (error, conn) => {
       if (error.code === SERVICE_UNAVAILABLE || error.code === SESSION_EXPIRED) {
         if (conn) {
           this._forget(conn.url)
@@ -132,7 +132,7 @@ class RoutingDriver extends Driver {
     // error transformer here is a no-op unlike the one in a regular session, this is so because errors are
     // handled in the rediscovery promise chain and we do not need to do this in the error transformer
     const errorTransformer = error => error;
-    return new RoutingSession(connectionPromise, this._releaseConnection(connectionPromise), errorTransformer);
+    return new RoutingSession(connectionPromise, errorTransformer);
   }
 
   _forget(url) {
@@ -160,8 +160,8 @@ class RoutingDriver extends Driver {
 }
 
 class RoutingSession extends Session {
-  constructor(connectionPromise, onClose, onFailedConnection) {
-    super(connectionPromise, onClose);
+  constructor(connectionPromise, onFailedConnection) {
+    super(connectionPromise);
     this._onFailedConnection = onFailedConnection;
   }
 

src/v1/session.js

@@ -16,11 +16,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import StreamObserver from "./internal/stream-observer";
-import Result from "./result";
-import Transaction from "./transaction";
-import {newError} from "./error";
-import {assertString} from "./internal/util";
+import StreamObserver from './internal/stream-observer';
+import Result from './result';
+import Transaction from './transaction';
+import {newError} from './error';
+import {assertString} from './internal/util';
 
 /**
   * A Session instance is used for handling the connection and
@@ -32,11 +32,10 @@ class Session {
   /**
    * @constructor
    * @param {Promise.<Connection>} connectionPromise - Promise of a connection to use
-   * @param {function()} onClose - Function to be called on connection close
    */
-  constructor( connectionPromise, onClose ) {
+  constructor(connectionPromise) {
     this._connectionPromise = connectionPromise;
-    this._onClose = onClose;
+    this._open = true;
     this._hasTx = false;
   }
 
@@ -102,25 +101,41 @@ class Session {
 
   /**
    * Close this session.
-   * @param {function()} cb - Function to be called after the session has been closed
+   * @param {function()} callback - Function to be called after the session has been closed
    * @return
    */
-  close(cb=(()=>null)) {
-    if(this._onClose) {
-      try {
-        this._onClose(cb);
-      } finally {
-        this._onClose = null;
-      }
+  close(callback = (() => null)) {
+    if (this._open) {
+      this._open = false;
+      this._releaseCurrentConnection().then(callback);
     } else {
-      cb();
+      callback();
     }
   }
 
   //Can be overridden to add error callback on RUN
   _onRunFailure() {
     return (err) => {return err};
   }
+
+  /**
+   * Return the current pooled connection instance to the connection pool.
+   * We don't pool Session instances, to avoid users using the Session after they've called close.
+   * The `Session` object is just a thin wrapper around Connection anyway, so it makes little difference.
+   * @return {Promise} - promise resolved then connection is returned to the pool.
+   * @private
+   */
+  _releaseCurrentConnection() {
+    return this._connectionPromise.then(conn => {
+      // Queue up a 'reset', to ensure the next user gets a clean session to work with.
+      conn.reset();
+      conn.sync();
+
+      // Return connection to the pool
+      conn._release();
+    }).catch(ignoredError => {
+    });
+  }
 }
 
 /** Internal stream observer used for transactional results*/

test/v1/session.test.js

@@ -45,16 +45,56 @@ describe('session', () => {
     driver.close();
   });
 
-  it('close should be idempotent ', () => {
-    // Given
-    let counter = 0;
-    const _session = new Session(null, () => {
-      counter++;
+  it('close should invoke callback ', done => {
+    const connection = new FakeConnection();
+    const session = new Session(Promise.resolve(connection));
+
+    session.close(done);
+  });
+
+  it('close should invoke callback even when already closed ', done => {
+    const connection = new FakeConnection();
+    const session = new Session(Promise.resolve(connection));
+
+    session.close(() => {
+      session.close(() => {
+        session.close(() => {
+          done();
+        });
+      });
+    });
+  });
+
+  it('close should be idempotent ', done => {
+    const connection = new FakeConnection();
+    const session = new Session(Promise.resolve(connection));
+
+    session.close(() => {
+      expect(connection.closedOnce()).toBeTruthy();
+
+      session.close(() => {
+        expect(connection.closedOnce()).toBeTruthy();
+
+        session.close(() => {
+          expect(connection.closedOnce()).toBeTruthy();
+          done();
+        });
+      });
+    });
+  });
+
+  it('should be possible to close driver after closing session with failed tx ', done => {
+    const driver = neo4j.driver('bolt://localhost', neo4j.auth.basic('neo4j', 'neo4j'));
+    const session = driver.session();
+    const tx = session.beginTransaction();
+    tx.run('INVALID QUERY').catch(() => {
+      tx.rollback().catch(() => {
+        session.close(() => {
+          driver.close();
+          done();
+        });
+      });
     });
-    _session.close();
-    expect(counter).toBe(1);
-    _session.close();
-    expect(counter).toBe(1);
   });
 
   it('should expose basic run/subscribe ', done => {
@@ -74,7 +114,7 @@ describe('session', () => {
     });
   });
 
-  it('should keep context in subscribe methods ', function (done) {
+  it('should keep context in subscribe methods ', done => {
     // Given
     function MyObserver() {
       this.local = 'hello';
@@ -87,7 +127,7 @@ describe('session', () => {
         expect(privateLocal).toBe('hello');
         expect(this.local).toBe('hello');
         done();
-      }
+      };
     }
 
     // When & Then
@@ -400,6 +440,31 @@ describe('session', () => {
   it('should fail for illegal session mode', () => {
     expect(() => driver.session('ILLEGAL_MODE')).toThrow();
   });
+
+  class FakeConnection {
+
+    constructor() {
+      this.resetInvoked = 0;
+      this.syncInvoked = 0;
+      this.releaseInvoked = 0;
+    }
+
+    reset() {
+      this.resetInvoked++;
+    }
+
+    sync() {
+      this.syncInvoked++;
+    }
+
+    _release() {
+      this.releaseInvoked++;
+    }
+
+    closedOnce() {
+      return this.resetInvoked === 1 && this.syncInvoked === 1 && this.releaseInvoked === 1;
+    }
+  }
 });