Make session aware of connection provider

So it could later use it to obtain connections on demand rather than
hold a single connection permanently.

Author: lutovich

src/v1/driver.js

@@ -58,7 +58,7 @@ class Driver {
       Driver._validateConnection.bind(this),
       config.connectionPoolSize
     );
-    this._connectionProvider = this._createConnectionProvider(url, this._pool);
+    this._connectionProvider = this._createConnectionProvider(url, this._pool, this._driverOnErrorCallback.bind(this));
   }
 
   /**
@@ -113,15 +113,7 @@ class Driver {
    */
   session(mode) {
     const sessionMode = Driver._validateSessionMode(mode);
-    const connectionPromise = this._connectionProvider.acquireConnection(sessionMode);
-    connectionPromise.catch((err) => {
-      if (this.onError && err.code === SERVICE_UNAVAILABLE) {
-        this.onError(err);
-      } else {
-        //we don't need to tell the driver about this error
-      }
-    });
-    return this._createSession(connectionPromise);
+    return this._createSession(sessionMode, this._connectionProvider);
   }
 
   static _validateSessionMode(rawMode) {
@@ -133,13 +125,22 @@ class Driver {
   }
 
   //Extension point
-  _createConnectionProvider(address, connectionPool) {
-    return new DirectConnectionProvider(address, connectionPool);
+  _createConnectionProvider(address, connectionPool, driverOnErrorCallback) {
+    return new DirectConnectionProvider(address, connectionPool, driverOnErrorCallback);
   }
 
   //Extension point
-  _createSession(connectionPromise) {
-    return new Session(connectionPromise);
+  _createSession(mode, connectionProvider) {
+    return new Session(mode, connectionProvider);
+  }
+
+  _driverOnErrorCallback(error) {
+    const userDefinedOnErrorCallback = this.onError;
+    if (userDefinedOnErrorCallback && error.code === SERVICE_UNAVAILABLE) {
+      userDefinedOnErrorCallback(error);
+    } else {
+      // we don't need to tell the driver about this error
+    }
   }
 
   /**

src/v1/internal/connection-providers.js

@@ -29,32 +29,43 @@ class ConnectionProvider {
   acquireConnection(mode) {
     throw new Error('Abstract method');
   }
+
+  _withAdditionalOnErrorCallback(connectionPromise, driverOnErrorCallback) {
+    // install error handler from the driver on the connection promise; this callback is installed separately
+    // so that it does not handle errors, instead it is just an additional error reporting facility.
+    connectionPromise.catch(error => driverOnErrorCallback(error));
+    // return the original connection promise
+    return connectionPromise;
+  }
 }
 
 export class DirectConnectionProvider extends ConnectionProvider {
 
-  constructor(address, connectionPool) {
+  constructor(address, connectionPool, driverOnErrorCallback) {
     super();
     this._address = address;
     this._connectionPool = connectionPool;
+    this._driverOnErrorCallback = driverOnErrorCallback;
   }
 
   acquireConnection(mode) {
-    return Promise.resolve(this._connectionPool.acquire(this._address));
+    const connectionPromise = Promise.resolve(this._connectionPool.acquire(this._address));
+    return this._withAdditionalOnErrorCallback(connectionPromise, this._driverOnErrorCallback);
   }
 }
 
 export class LoadBalancer extends ConnectionProvider {
 
-  constructor(address, connectionPool) {
+  constructor(address, connectionPool, driverOnErrorCallback) {
     super();
     this._routingTable = new RoutingTable(new RoundRobinArray([address]));
     this._rediscovery = new Rediscovery();
     this._connectionPool = connectionPool;
+    this._driverOnErrorCallback = driverOnErrorCallback;
   }
 
   acquireConnection(mode) {
-    return this._freshRoutingTable().then(routingTable => {
+    const connectionPromise = this._freshRoutingTable().then(routingTable => {
       if (mode === READ) {
         return this._acquireConnectionToServer(routingTable.readers, 'read');
       } else if (mode === WRITE) {
@@ -63,6 +74,7 @@ export class LoadBalancer extends ConnectionProvider {
         throw newError('Illegal mode ' + mode);
       }
     });
+    return this._withAdditionalOnErrorCallback(connectionPromise, this._driverOnErrorCallback);
   }
 
   forget(address) {
@@ -132,7 +144,8 @@ export class LoadBalancer extends ConnectionProvider {
   _createSessionForRediscovery(routerAddress) {
     const connection = this._connectionPool.acquire(routerAddress);
     const connectionPromise = Promise.resolve(connection);
-    return new Session(connectionPromise);
+    const connectionProvider = new SingleConnectionProvider(connectionPromise);
+    return new Session(READ, connectionProvider);
   }
 
   _updateRoutingTable(newRoutingTable) {
@@ -153,3 +166,17 @@ export class LoadBalancer extends ConnectionProvider {
     }
   }
 }
+
+export class SingleConnectionProvider extends ConnectionProvider {
+
+  constructor(connectionPromise) {
+    super();
+    this._connectionPromise = connectionPromise;
+  }
+
+  acquireConnection(mode) {
+    const connectionPromise = this._connectionPromise;
+    this._connectionPromise = null;
+    return connectionPromise;
+  }
+}

src/v1/routing-driver.js

@@ -31,30 +31,24 @@ class RoutingDriver extends Driver {
     super(url, userAgent, token, RoutingDriver._validateConfig(config));
   }
 
-  _createConnectionProvider(address, connectionPool) {
-    return new LoadBalancer(address, connectionPool);
+  _createConnectionProvider(address, connectionPool, driverOnErrorCallback) {
+    return new LoadBalancer(address, connectionPool, driverOnErrorCallback);
   }
 
-  _createSession(connectionPromise) {
-    return new RoutingSession(connectionPromise, (error, conn) => {
+  _createSession(mode, connectionProvider) {
+    return new RoutingSession(mode, connectionProvider, (error, conn) => {
       if (error.code === SERVICE_UNAVAILABLE || error.code === SESSION_EXPIRED) {
+        // connection is undefined if error happened before connection was acquired
         if (conn) {
           this._connectionProvider.forget(conn.url);
-        } else {
-          connectionPromise.then((conn) => {
-            this._connectionProvider.forget(conn.url);
-          }).catch(() => {/*ignore*/});
         }
         return error;
       } else if (error.code === 'Neo.ClientError.Cluster.NotALeader') {
         let url = 'UNKNOWN';
+        // connection is undefined if error happened before connection was acquired
         if (conn) {
           url = conn.url;
           this._connectionProvider.forgetWriter(conn.url);
-        } else {
-          connectionPromise.then((conn) => {
-            this._connectionProvider.forgetWriter(conn.url);
-          }).catch(() => {/*ignore*/});
         }
         return newError('No longer possible to write to server at ' + url, SESSION_EXPIRED);
       } else {
@@ -72,8 +66,8 @@ class RoutingDriver extends Driver {
 }
 
 class RoutingSession extends Session {
-  constructor(connectionPromise, onFailedConnection) {
-    super(connectionPromise);
+  constructor(mode, connectionProvider, onFailedConnection) {
+    super(mode, connectionProvider);
     this._onFailedConnection = onFailedConnection;
   }
 

src/v1/session.js

@@ -31,10 +31,12 @@ import {assertString} from './internal/util';
 class Session {
   /**
    * @constructor
-   * @param {Promise.<Connection>} connectionPromise - Promise of a connection to use
+   * todo: doc params
    */
-  constructor(connectionPromise) {
-    this._connectionPromise = connectionPromise;
+  constructor(mode, connectionProvider) {
+    this._mode = mode;
+    this._connectionProvider = connectionProvider;
+    this._connectionPromise = this._connectionProvider.acquireConnection(this._mode);
     this._open = true;
     this._hasTx = false;
   }

test/internal/connection-providers.test.js

@@ -25,11 +25,14 @@ import RoundRobinArray from '../../src/v1/internal/round-robin-array';
 import {DirectConnectionProvider, LoadBalancer} from '../../src/v1/internal/connection-providers';
 import Pool from '../../src/v1/internal/pool';
 
+const NO_OP_DRIVER_CALLBACK = () => {
+};
+
 describe('DirectConnectionProvider', () => {
 
   it('acquires connection from the pool', done => {
     const pool = newPool();
-    const connectionProvider = new DirectConnectionProvider('localhost:123', pool);
+    const connectionProvider = newDirectConnectionProvider('localhost:123', pool);
 
     connectionProvider.acquireConnection(READ).then(connection => {
       expect(connection).toBeDefined();
@@ -132,7 +135,7 @@ describe('LoadBalancer', () => {
   });
 
   it('initializes routing table with the given router', () => {
-    const loadBalancer = new LoadBalancer('server-ABC', newPool());
+    const loadBalancer = new LoadBalancer('server-ABC', newPool(), NO_OP_DRIVER_CALLBACK);
 
     expectRoutingTable(loadBalancer,
       ['server-ABC'],
@@ -581,8 +584,12 @@ describe('LoadBalancer', () => {
 
 });
 
+function newDirectConnectionProvider(address, pool) {
+  return new DirectConnectionProvider(address, pool, NO_OP_DRIVER_CALLBACK);
+}
+
 function newLoadBalancer(routers, readers, writers, pool = null, expirationTime = Integer.MAX_VALUE, routerToRoutingTable = {}) {
-  const loadBalancer = new LoadBalancer(null, pool || newPool());
+  const loadBalancer = new LoadBalancer(null, pool || newPool(), NO_OP_DRIVER_CALLBACK);
   loadBalancer._routingTable = new RoutingTable(
     new RoundRobinArray(routers),
     new RoundRobinArray(readers),

test/v1/session.test.js

@@ -20,6 +20,8 @@
 import neo4j from '../../src/v1';
 import {statementType} from '../../src/v1/result-summary';
 import Session from '../../src/v1/session';
+import {READ} from '../../src/v1/driver';
+import {SingleConnectionProvider} from '../../src/v1/internal/connection-providers';
 
 describe('session', () => {
 
@@ -47,14 +49,14 @@ describe('session', () => {
 
   it('close should invoke callback ', done => {
     const connection = new FakeConnection();
-    const session = new Session(Promise.resolve(connection));
+    const session = newSessionWithConnection(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));
+    const session = newSessionWithConnection(connection);
 
     session.close(() => {
       session.close(() => {
@@ -67,7 +69,7 @@ describe('session', () => {
 
   it('close should be idempotent ', done => {
     const connection = new FakeConnection();
-    const session = new Session(Promise.resolve(connection));
+    const session = newSessionWithConnection(connection);
 
     session.close(() => {
       expect(connection.closedOnce()).toBeTruthy();
@@ -441,6 +443,11 @@ describe('session', () => {
     expect(() => driver.session('ILLEGAL_MODE')).toThrow();
   });
 
+  function newSessionWithConnection(connection) {
+    const connectionProvider = new SingleConnectionProvider(Promise.resolve(connection));
+    return new Session(READ, connectionProvider);
+  }
+
   class FakeConnection {
 
     constructor() {
@@ -449,6 +456,12 @@ describe('session', () => {
       this.releaseInvoked = 0;
     }
 
+    run() {
+    }
+
+    discardAll() {
+    }
+
     reset() {
       this.resetInvoked++;
     }