Routing context in Bolt URI

Commit makes it possible to specify routing parameters in Bolt URI query.
These parameters are then send to server when doing rediscovery. This
is true only for 3.2+ Neo4j database because it supports
`getRoutingTable(context)` procedure.

Renamed `GetServersUtil` to `RoutingUtil` because it now knows about both
routing procedures `getServers` and `getRoutingTable`.

Author: lutovich

src/v1/driver.js

@@ -58,7 +58,13 @@ class Driver {
       Driver._validateConnection.bind(this),
       config.connectionPoolSize
     );
-    this._connectionProvider = this._createConnectionProvider(url, this._pool, this._driverOnErrorCallback.bind(this));
+
+    /**
+     * Reference to the connection provider. Initialized lazily by {@link _getOrCreateConnectionProvider}.
+     * @type {ConnectionProvider}
+     * @private
+     */
+    this._connectionProvider = null;
   }
 
   /**
@@ -115,7 +121,8 @@ class Driver {
    */
   session(mode, bookmark) {
     const sessionMode = Driver._validateSessionMode(mode);
-    return this._createSession(sessionMode, this._connectionProvider, bookmark, this._config);
+    const connectionProvider = this._getOrCreateConnectionProvider();
+    return this._createSession(sessionMode, connectionProvider, bookmark, this._config);
   }
 
   static _validateSessionMode(rawMode) {
@@ -142,6 +149,14 @@ class Driver {
     return SERVICE_UNAVAILABLE;
   }
 
+  _getOrCreateConnectionProvider() {
+    if (!this._connectionProvider) {
+      const driverOnErrorCallback = this._driverOnErrorCallback.bind(this);
+      this._connectionProvider = this._createConnectionProvider(this._url, this._pool, driverOnErrorCallback);
+    }
+    return this._connectionProvider;
+  }
+
   _driverOnErrorCallback(error) {
     const userDefinedOnErrorCallback = this.onError;
     if (userDefinedOnErrorCallback && error.code === SERVICE_UNAVAILABLE) {
@@ -189,8 +204,9 @@ class _ConnectionStreamObserver extends StreamObserver {
     if (this._driver.onCompleted) {
       this._driver.onCompleted(message);
     }
-    if (this._conn && message && message.server) {
-      this._conn.setServerVersion(message.server);
+
+    if (this._observer && this._observer.onComplete) {
+      this._observer.onCompleted(message);
     }
   }
 }

src/v1/index.js

@@ -128,20 +128,18 @@ function driver(url, authToken, config = {}) {
   assertString(url, 'Bolt URL');
   const scheme = parseScheme(url);
   const routingContext = parseRoutingContext(url);
-  if (scheme === "bolt+routing://") {
+  if (scheme === 'bolt+routing://') {
     return new RoutingDriver(parseUrl(url), routingContext, USER_AGENT, authToken, config);
-  } else if (scheme === "bolt://") {
-    if(!isEmptyObjectOrNull(routingContext))
-    {
-      throw new Error("Routing context are not supported with scheme 'bolt'. Given URI: '" + url + "'");
+  } else if (scheme === 'bolt://') {
+    if (!isEmptyObjectOrNull(routingContext)) {
+      throw new Error(`Routing parameters are not supported with scheme 'bolt'. Given URL: '${url}'`);
     }
     return new Driver(parseUrl(url), USER_AGENT, authToken, config);
   } else {
-    throw new Error("Unknown scheme: " + scheme);
+    throw new Error(`Unknown scheme: ${scheme}`);
   }
 }
 
-
 const types ={
   Node,
   Relationship,

src/v1/internal/connection-providers.js

@@ -25,7 +25,7 @@ import RoutingTable from './routing-table';
 import Rediscovery from './rediscovery';
 import hasFeature from './features';
 import {DnsHostNameResolver, DummyHostNameResolver} from './host-name-resolvers';
-import GetServersUtil from './get-servers-util';
+import RoutingUtil from './routing-util';
 
 class ConnectionProvider {
 
@@ -66,7 +66,7 @@ export class LoadBalancer extends ConnectionProvider {
     super();
     this._seedRouter = address;
     this._routingTable = new RoutingTable(new RoundRobinArray([this._seedRouter]));
-    this._rediscovery = new Rediscovery(new GetServersUtil(routingContext));
+    this._rediscovery = new Rediscovery(new RoutingUtil(routingContext));
     this._connectionPool = connectionPool;
     this._driverOnErrorCallback = driverOnErrorCallback;
     this._hostNameResolver = LoadBalancer._createHostNameResolver();
@@ -172,8 +172,10 @@ export class LoadBalancer extends ConnectionProvider {
 
   _createSessionForRediscovery(routerAddress) {
     const connection = this._connectionPool.acquire(routerAddress);
-    const connectionPromise = Promise.resolve(connection);
-    const connectionProvider = new SingleConnectionProvider(connectionPromise);
+    // initialized connection is required for routing procedure call
+    // server version needs to be known to decide which routing procedure to use
+    const initializedConnectionPromise = connection.initializationCompleted();
+    const connectionProvider = new SingleConnectionProvider(initializedConnectionPromise);
     return new Session(READ, connectionProvider);
   }
 

src/v1/internal/connector.js

@@ -26,6 +26,7 @@ import {Node, Path, PathSegment, Relationship, UnboundRelationship} from '../gra
 import {newError} from './../error';
 import ChannelConfig from './ch-config';
 import {parseHost, parsePort} from './util';
+import StreamObserver from './stream-observer';
 
 let Channel;
 if( NodeChannel.available ) {
@@ -183,6 +184,8 @@ class Connection {
     this._isHandlingFailure = false;
     this._currentFailure = null;
 
+    this._state = new ConnectionState(this);
+
     // Set to true on fatal errors, to get this out of session pool.
     this._isBroken = false;
 
@@ -330,7 +333,8 @@ class Connection {
   /** Queue an INIT-message to be sent to the database */
   initialize( clientName, token, observer ) {
     log("C", "INIT", clientName, token);
-    this._queueObserver(observer);
+    const initObserver = this._state.wrap(observer);
+    this._queueObserver(initObserver);
     this._packer.packStruct( INIT, [this._packable(clientName), this._packable(token)],
       (err) => this._handleFatalError(err) );
     this._chunker.messageBoundary();
@@ -416,6 +420,15 @@ class Connection {
     }
   }
 
+  /**
+   * Get promise resolved when connection initialization succeed or rejected when it fails.
+   * Connection is initialized using {@link initialize} function.
+   * @return {Promise<Connection>} the result of connection initialization.
+   */
+  initializationCompleted() {
+    return this._state.initializationCompleted();
+  }
+
   /**
    * Synchronize - flush all queued outgoing messages and route their responses
    * to their respective handlers.
@@ -450,6 +463,59 @@ class Connection {
   }
 }
 
+class ConnectionState {
+
+  /**
+   * @constructor
+   * @param {Connection} connection the connection to track state for.
+   */
+  constructor(connection) {
+    this._connection = connection;
+    this._resolvePromise = null;
+    this._promise = new Promise(resolve => {
+      this._resolvePromise = resolve;
+    });
+  }
+
+  /**
+   * Wrap the given observer to track connection's initialization state.
+   * @param {StreamObserver} observer the observer used for INIT message.
+   * @return {StreamObserver} updated observer.
+   */
+  wrap(observer) {
+    return {
+      onNext: record => {
+        if (observer && observer.onNext) {
+          observer.onNext(record);
+        }
+      },
+      onError: error => {
+        this._resolvePromise(Promise.reject(error));
+        if (observer && observer.onError) {
+          observer.onError(error);
+        }
+      },
+      onCompleted: metaData => {
+        if (metaData && metaData.server) {
+          this._connection.setServerVersion(metaData.server);
+        }
+        this._resolvePromise(this._connection);
+        if (observer && observer.onCompleted) {
+          observer.onCompleted(metaData);
+        }
+      }
+    };
+  }
+
+  /**
+   * Get promise resolved when connection initialization succeed or rejected when it fails.
+   * @return {Promise<Connection>} the result of connection initialization.
+   */
+  initializationCompleted() {
+    return this._promise;
+  }
+}
+
 /**
  * Crete new connection to the provided url.
  * @access private

src/v1/internal/rediscovery.js

@@ -17,18 +17,27 @@
  * limitations under the License.
  */
 
-import GetServersUtil from "./get-servers-util";
-import RoutingTable from "./routing-table";
-import {newError, PROTOCOL_ERROR} from "../error";
+import RoutingTable from './routing-table';
+import {newError, PROTOCOL_ERROR} from '../error';
 
 export default class Rediscovery {
 
-  constructor(getServersUtil) {
-    this._getServersUtil = getServersUtil;
+  /**
+   * @constructor
+   * @param {RoutingUtil} routingUtil the util to use.
+   */
+  constructor(routingUtil) {
+    this._routingUtil = routingUtil;
   }
 
+  /**
+   * Try to fetch new routing table from the given router.
+   * @param {Session} session the session to use.
+   * @param {string} routerAddress the URL of the router.
+   * @return {Promise<RoutingTable>} promise resolved with new routing table or null when connection error happened.
+   */
   lookupRoutingTableOnRouter(session, routerAddress) {
-    return this._getServersUtil.callGetServers(session, routerAddress).then(records => {
+    return this._routingUtil.callRoutingProcedure(session, routerAddress).then(records => {
       if (records === null) {
         // connection error happened, unable to retrieve routing table from this router, next one should be queried
         return null;
@@ -42,8 +51,8 @@ export default class Rediscovery {
 
       const record = records[0];
 
-      const expirationTime = this._getServersUtil.parseTtl(record, routerAddress);
-      const {routers, readers, writers} = this._getServersUtil.parseServers(record, routerAddress);
+      const expirationTime = this._routingUtil.parseTtl(record, routerAddress);
+      const {routers, readers, writers} = this._routingUtil.parseServers(record, routerAddress);
 
       Rediscovery._assertNonEmpty(routers, 'routers', routerAddress);
       Rediscovery._assertNonEmpty(readers, 'readers', routerAddress);

src/v1/internal/get-servers-util.js renamed to src/v1/internal/routing-util.js

@@ -20,44 +20,39 @@
 import RoundRobinArray from './round-robin-array';
 import {newError, PROTOCOL_ERROR, SERVICE_UNAVAILABLE} from '../error';
 import Integer, {int} from '../integer';
-import {ServerVersion, VERSION3_2} from './server-version-util'
+import {ServerVersion, VERSION_3_2_0} from './server-version';
 
 const CALL_GET_SERVERS = 'CALL dbms.cluster.routing.getServers';
-const GET_ROUTING_TABLE_PARAM = "context";
-const CALL_GET_ROUTING_TABLE = "CALL dbms.cluster.routing.getRoutingTable({"
-  + GET_ROUTING_TABLE_PARAM + "})";
+const GET_ROUTING_TABLE_PARAM = 'context';
+const CALL_GET_ROUTING_TABLE = 'CALL dbms.cluster.routing.getRoutingTable({' + GET_ROUTING_TABLE_PARAM + '})';
 const PROCEDURE_NOT_FOUND_CODE = 'Neo.ClientError.Procedure.ProcedureNotFound';
 
-export default class GetServersUtil {
+export default class RoutingUtil {
 
-  constructor(routingContext={}) {
+  constructor(routingContext) {
     this._routingContext = routingContext;
   }
 
-  callGetServers(session, routerAddress) {
-    session.run("RETURN 1").then(result=>{
-      let statement = {text:CALL_GET_SERVERS};
-
-      if(ServerVersion.fromString(result.summary.server.version).compare(VERSION3_2)>=0)
-      {
-        statement = {
-          text:CALL_GET_ROUTING_TABLE,
-          parameters:{GET_ROUTING_TABLE_PARAM: this._routingContext}};
+  /**
+   * Invoke routing procedure using the given session.
+   * @param {Session} session the session to use.
+   * @param {string} routerAddress the URL of the router.
+   * @return {Promise<Record[]>} promise resolved with records returned by the procedure call or null if
+   * connection error happened.
+   */
+  callRoutingProcedure(session, routerAddress) {
+    return this._callAvailableRoutingProcedure(session).then(result => {
+      session.close();
+      return result.records;
+    }).catch(error => {
+      if (error.code === PROCEDURE_NOT_FOUND_CODE) {
+        // throw when getServers procedure not found because this is clearly a configuration issue
+        throw newError('Server ' + routerAddress + ' could not perform routing. ' +
+          'Make sure you are connecting to a causal cluster', SERVICE_UNAVAILABLE);
       }
-
-      return session.run(statement).then(result => {
-        session.close();
-        return result.records;
-      }).catch(error => {
-        if (error.code === PROCEDURE_NOT_FOUND_CODE) {
-          // throw when getServers procedure not found because this is clearly a configuration issue
-          throw newError('Server ' + routerAddress + ' could not perform routing. ' +
-            'Make sure you are connecting to a causal cluster', SERVICE_UNAVAILABLE);
-        }
-        // return nothing when failed to connect because code higher in the callstack is still able to retry with a
-        // different session towards a different router
-        return null;
-      });
+      // return nothing when failed to connect because code higher in the callstack is still able to retry with a
+      // different session towards a different router
+      return null;
     });
   }
 
@@ -111,4 +106,18 @@ export default class GetServersUtil {
         PROTOCOL_ERROR);
     }
   }
+
+  _callAvailableRoutingProcedure(session) {
+    return session._run(null, null, (connection, streamObserver) => {
+      const serverVersionString = connection.server.version;
+      const serverVersion = ServerVersion.fromString(serverVersionString);
+
+      if (serverVersion.compareTo(VERSION_3_2_0) >= 0) {
+        const params = {[GET_ROUTING_TABLE_PARAM]: this._routingContext};
+        connection.run(CALL_GET_ROUTING_TABLE, params, streamObserver);
+      } else {
+        connection.run(CALL_GET_SERVERS, {}, streamObserver);
+      }
+    });
+  }
 }

src/v1/session.js

@@ -59,28 +59,34 @@ class Session {
    * @return {Result} - New Result
    */
   run(statement, parameters = {}) {
-    if(typeof statement === 'object' && statement.text) {
+    if (typeof statement === 'object' && statement.text) {
       parameters = statement.parameters || {};
       statement = statement.text;
     }
-    assertString(statement, "Cypher statement");
+    assertString(statement, 'Cypher statement');
 
+    return this._run(statement, parameters, (connection, streamObserver) =>
+      connection.run(statement, parameters, streamObserver)
+    );
+  }
+
+  _run(statement, parameters, statementRunner) {
     const streamObserver = new _RunObserver(this._onRunFailure());
     const connectionHolder = this._connectionHolderWithMode(this._mode);
     if (!this._hasTx) {
       connectionHolder.initializeConnection();
       connectionHolder.getConnection().then(connection => {
         streamObserver.resolveConnection(connection);
-        connection.run(statement, parameters, streamObserver);
+        statementRunner(connection, streamObserver);
         connection.pullAll(streamObserver);
         connection.sync();
       }).catch(error => streamObserver.onError(error));
     } else {
-      streamObserver.onError(newError("Statements cannot be run directly on a "
-       + "session with an open transaction; either run from within the "
-       + "transaction or use a different session."));
+      streamObserver.onError(newError('Statements cannot be run directly on a ' +
+        'session with an open transaction; either run from within the ' +
+        'transaction or use a different session.'));
     }
-    return new Result( streamObserver, statement, parameters, () => streamObserver.meta(), connectionHolder );
+    return new Result(streamObserver, statement, parameters, () => streamObserver.meta(), connectionHolder);
   }
 
   /**

test/internal/connection-providers.test.js

@@ -1102,6 +1102,10 @@ class FakeConnection {
   static create(address, release) {
     return new FakeConnection(address, release);
   }
+
+  initializationCompleted() {
+    return Promise.resolve(this);
+  }
 }
 
 class FakeRediscovery {