Split routing logic into multiple components

Previously it was mostly located in routing driver which owned the cluster
view and all logic to call `getServers` and interpret it's output/failures.

This introduces two new internal components:
 * `GetServersUtil` - responsible for calling `getServers` procedure and
 processing it's output
 * `Rediscovery` - responsible for fetching a routing table from the specified
 server

Routing driver now owns a routing table and an instance of rediscovery. It
uses the later one to update the routing table when needed.

Added dev dependency to lolex(https://github.com/sinonjs/lolex) to be able to
mock system time and test things better.

Author: lutovich

package.json

@@ -49,6 +49,7 @@
     "gulp-util": "^3.0.6",
     "gulp-watch": "^4.3.5",
     "jasmine-reporters": "^2.0.7",
+    "lolex": "^1.5.2",
     "merge-stream": "^1.0.0",
     "minimist": "^1.2.0",
     "phantomjs-prebuilt": "^2.1.7 ",

src/v1/driver.js

@@ -24,7 +24,7 @@ import {connect} from "./internal/connector";
 import StreamObserver from './internal/stream-observer';
 import {newError, SERVICE_UNAVAILABLE} from "./error";
 
-let READ = 'READ', WRITE = 'WRITE';
+const READ = 'READ', WRITE = 'WRITE';
 /**
  * A driver maintains one or more {@link Session sessions} with a remote
  * Neo4j instance. Through the {@link Session sessions} you can send statements

src/v1/error.js

@@ -20,8 +20,10 @@
 // A common place for constructing error objects, to keep them
 // uniform across the driver surface.
 
-let SERVICE_UNAVAILABLE = 'ServiceUnavailable';
-let SESSION_EXPIRED = 'SessionExpired';
+const SERVICE_UNAVAILABLE = 'ServiceUnavailable';
+const SESSION_EXPIRED = 'SessionExpired';
+const PROTOCOL_ERROR = 'ProtocolError';
+
 function newError(message, code="N/A") {
   // TODO: Idea is that we can check the code here and throw sub-classes
   // of Neo4jError as appropriate
@@ -40,5 +42,6 @@ export {
   newError,
   Neo4jError,
   SERVICE_UNAVAILABLE,
-  SESSION_EXPIRED
+  SESSION_EXPIRED,
+  PROTOCOL_ERROR
 }

src/v1/internal/get-servers-util.js

@@ -0,0 +1,94 @@
+/**
+ * Copyright (c) 2002-2017 "Neo Technology,","
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import RoundRobinArray from "./round-robin-array";
+import {newError, PROTOCOL_ERROR, SERVICE_UNAVAILABLE} from "../error";
+import Integer, {int} from "../integer";
+
+const PROCEDURE_CALL = 'CALL dbms.cluster.routing.getServers';
+const PROCEDURE_NOT_FOUND_CODE = 'Neo.ClientError.Procedure.ProcedureNotFound';
+
+export default class GetServersUtil {
+
+  callGetServers(session, routerAddress) {
+    return session.run(PROCEDURE_CALL).then(result => {
+      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;
+    });
+  }
+
+  parseTtl(record, routerAddress) {
+    try {
+      const now = int(Date.now());
+      const expires = record.get('ttl').multiply(1000).add(now);
+      // if the server uses a really big expire time like Long.MAX_VALUE this may have overflowed
+      if (expires.lessThan(now)) {
+        return Integer.MAX_VALUE;
+      }
+      return expires;
+    } catch (error) {
+      throw newError(
+        'Unable to parse TTL entry from router ' + routerAddress + ' from record:\n' + JSON.stringify(record),
+        PROTOCOL_ERROR);
+    }
+  }
+
+  parseServers(record, routerAddress) {
+    try {
+      const servers = record.get('servers');
+
+      const routers = new RoundRobinArray();
+      const readers = new RoundRobinArray();
+      const writers = new RoundRobinArray();
+
+      servers.forEach(server => {
+        const role = server['role'];
+        const addresses = server['addresses'];
+
+        if (role === 'ROUTE') {
+          routers.pushAll(addresses);
+        } else if (role === 'WRITE') {
+          writers.pushAll(addresses);
+        } else if (role === 'READ') {
+          readers.pushAll(addresses);
+        } else {
+          throw newError('Unknown server role "' + role + '"', PROTOCOL_ERROR);
+        }
+      });
+
+      return {
+        routers: routers,
+        readers: readers,
+        writers: writers
+      }
+    } catch (ignore) {
+      throw newError(
+        'Unable to parse servers entry from router ' + routerAddress + ' from record:\n' + JSON.stringify(record),
+        PROTOCOL_ERROR);
+    }
+  }
+}

src/v1/internal/rediscovery.js

@@ -0,0 +1,65 @@
+/**
+ * Copyright (c) 2002-2017 "Neo Technology,","
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import GetServersUtil from "./get-servers-util";
+import RoutingTable from "./routing-table";
+import {newError, PROTOCOL_ERROR} from "../error";
+
+export default class Rediscovery {
+
+  constructor(getServersUtil) {
+    this._getServersUtil = getServersUtil || new GetServersUtil();
+  }
+
+  lookupRoutingTableOnRouter(session, routerAddress) {
+    return this._getServersUtil.callGetServers(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;
+      }
+
+      if (records.length !== 1) {
+        throw newError('Illegal response from router "' + routerAddress + '". ' +
+          'Received ' + records.length + ' records but expected only one.\n' + JSON.stringify(records),
+          PROTOCOL_ERROR);
+      }
+
+      const record = records[0];
+
+      const expirationTime = this._getServersUtil.parseTtl(record, routerAddress);
+      const {routers, readers, writers} = this._getServersUtil.parseServers(record, routerAddress);
+
+      Rediscovery._assertNonEmpty(routers, 'routers', routerAddress);
+      Rediscovery._assertNonEmpty(readers, 'readers', routerAddress);
+
+      if (writers.isEmpty()) {
+        // retrieved routing table has no writers, next router should be queried
+        return null;
+      }
+
+      return new RoutingTable(routers, readers, writers, expirationTime);
+    });
+  }
+
+  static _assertNonEmpty(serversRoundRobinArray, serversName, routerAddress) {
+    if (serversRoundRobinArray.isEmpty()) {
+      throw newError('Received no ' + serversName + ' from router ' + routerAddress, PROTOCOL_ERROR);
+    }
+  }
+}

src/v1/internal/round-robin-array.js

@@ -20,7 +20,8 @@
 /**
  * An array that lets you hop through the elements endlessly.
  */
-class RoundRobinArray {
+export default class RoundRobinArray {
+
   constructor(items) {
     this._items = items || [];
     this._offset = 0;
@@ -36,6 +37,10 @@ class RoundRobinArray {
   }
 
   pushAll(elems) {
+    if (!Array.isArray(elems)) {
+      throw new TypeError('Array expected but got: ' + elems);
+    }
+
     Array.prototype.push.apply(this._items, elems);
   }
 
@@ -55,5 +60,3 @@ class RoundRobinArray {
     this._items = this._items.filter(element => element !== item);
   }
 }
-
-export default RoundRobinArray

src/v1/internal/routing-table.js

@@ -17,6 +17,7 @@
  * limitations under the License.
  */
 import {int} from "../integer";
+import RoundRobinArray from "./round-robin-array";
 
 const MIN_ROUTERS = 1;
 
@@ -26,8 +27,7 @@ export default class RoutingTable {
     this.routers = routers || new RoundRobinArray();
     this.readers = readers || new RoundRobinArray();
     this.writers = writers || new RoundRobinArray();
-
-    this._expirationTime = expirationTime || int(0);
+    this.expirationTime = expirationTime || int(0);
   }
 
   forget(address) {
@@ -46,7 +46,7 @@ export default class RoutingTable {
   }
 
   isStale() {
-    return this._expirationTime.lessThan(Date.now()) ||
+    return this.expirationTime.lessThan(Date.now()) ||
       this.routers.size() <= MIN_ROUTERS ||
       this.readers.isEmpty() ||
       this.writers.isEmpty();