Align routing to spec

package.json

@@ -49,12 +49,15 @@
     "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",
+    "mustache": "^2.3.0",
     "phantomjs-prebuilt": "^2.1.7 ",
     "run-sequence": "^1.1.4",
     "semver": "^5.3.0",
     "through2": "~2.0.0",
+    "tmp": "0.0.31",
     "vinyl-buffer": "^1.0.0",
     "vinyl-source-stream": "^1.1.0"
   },

src/v1/driver.js

@@ -17,14 +17,13 @@
  * limitations under the License.
  */
 
-import Session from './session';
-import Pool from './internal/pool';
-import Integer from './integer';
+import Session from "./session";
+import Pool from "./internal/pool";
 import {connect} from "./internal/connector";
-import StreamObserver from './internal/stream-observer';
+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
@@ -111,38 +110,50 @@ class Driver {
    * @return {Session} new session.
    */
   session(mode) {
-    let connectionPromise = this._acquireConnection(mode);
+    const sessionMode = Driver._validateSessionMode(mode);
+    const connectionPromise = this._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, (cb) => {
-      // This 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.
-
-      // Queue up a 'reset', to ensure the next user gets a clean
-      // session to work with.
+    return this._createSession(connectionPromise, this._releaseConnection(connectionPromise));
+  }
 
-      connectionPromise.then( (conn) => {
+  /**
+   * 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(*=)} - 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( () => {/*ignore errors here*/});
+      }).catch(ignoredError => {
+      });
 
-      // Call user callback
-      if (cb) {
-        cb();
+      if (userDefinedCallback) {
+        userDefinedCallback();
       }
-    });
+    };
+  }
+
+  static _validateSessionMode(rawMode) {
+    const mode = rawMode || WRITE;
+    if (mode !== READ && mode !== WRITE) {
+      throw newError('Illegal session mode ' + mode);
+    }
+    return mode;
   }
 
   //Extension point

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/index.js

@@ -19,7 +19,7 @@
 
 import {int, isInt, inSafeRange, toNumber, toString} from './integer';
 import {Node, Relationship, UnboundRelationship, PathSegment, Path} from './graph-types'
-import {Neo4jError, SERVICE_UNAVAILABLE, SESSION_EXPIRED} from './error';
+import {Neo4jError, SERVICE_UNAVAILABLE, SESSION_EXPIRED, PROTOCOL_ERROR} from './error';
 import Result from './result';
 import ResultSummary from './result-summary';
 import Record from './record';
@@ -138,7 +138,8 @@ const session = {
 };
 const error = {
   SERVICE_UNAVAILABLE,
-  SESSION_EXPIRED
+  SESSION_EXPIRED,
+  PROTOCOL_ERROR
 };
 const integer = {
   toNumber,

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

@@ -0,0 +1,107 @@
+/**
+ * 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 => {
+      session.close();
+      return result.records;
+    }).catch(error => {
+      if (this._isProcedureNotFoundError(error)) {
+        // 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);
+    }
+  }
+
+  _isProcedureNotFoundError(error) {
+    let errorCode = error.code;
+    if (!errorCode) {
+      try {
+        errorCode = error.fields[0].code;
+      } catch (e) {
+        errorCode = 'UNKNOWN';
+      }
+    }
+    return errorCode === PROCEDURE_NOT_FOUND_CODE;
+  }
+}

src/v1/internal/rediscovery.js

@@ -0,0 +1,62 @@
+/**
+ * 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);
+      // case with no writers is processed higher in the promise chain because only RoutingDriver knows
+      // how to deal with such table and how to treat router that returned such table
+
+      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,39 +20,34 @@
 /**
  * An array that lets you hop through the elements endlessly.
  */
-class RoundRobinArray {
+export default class RoundRobinArray {
+
   constructor(items) {
     this._items = items || [];
-    this._index = 0;
+    this._offset = 0;
   }
 
   next() {
-    let elem = this._items[this._index];
-    if (this._items.length === 0) {
-      this._index = 0;
-    } else {
-      this._index = (this._index + 1) % (this._items.length);
+    if (this.isEmpty()) {
+      return null;
     }
-    return elem;
-  }
-
-  push(elem) {
-    this._items.push(elem);
+    const index = this._offset % this.size();
+    this._offset++;
+    return this._items[index];
   }
 
   pushAll(elems) {
+    if (!Array.isArray(elems)) {
+      throw new TypeError('Array expected but got: ' + elems);
+    }
+
     Array.prototype.push.apply(this._items, elems);
   }
 
-  empty() {
+  isEmpty() {
     return this._items.length === 0;
   }
 
-  clear() {
-    this._items = [];
-    this._index = 0;
-  }
-
   size() {
     return this._items.length;
   }
@@ -62,21 +57,6 @@ class RoundRobinArray {
   }
 
   remove(item) {
-    let index = this._items.indexOf(item);
-    while (index != -1) {
-      this._items.splice(index, 1);
-      if (index < this._index) {
-        this._index -= 1;
-      }
-      //make sure we are in range
-      if (this._items.length === 0) {
-        this._index = 0;
-      } else {
-        this._index %= this._items.length;
-      }
-      index = this._items.indexOf(item, index);
-    }
+    this._items = this._items.filter(element => element !== item);
   }
 }
-
-export default RoundRobinArray