Refactor Session and Transaction maintains a promise

Instead of keeping a connection `Session` and `Transaction` now keep
a promise to a connection instead of a resolved connection.

Author: pontusmelke

src/v1/driver.js

@@ -116,7 +116,7 @@ class Driver {
   }
 
   _createSession(conn) {
-    return new Session(conn, (cb) => {
+    return new Session(new Promise((resolve, reject) => resolve(conn)), (cb) => {
       // This gets called on Session#close(), and is where we return
       // the pooled 'connection' instance.
 
@@ -204,6 +204,47 @@ class RoundRobinArray {
 }
 
 let GET_SERVERS = "CALL dbms.cluster.routing.getServers";
+
+class ClusterView {
+  constructor(expires, routers, readers, writers) {
+    this.expires = expires;
+    this.routers = routers;
+    this.readers = readers;
+    this.routers = writers;
+  }
+}
+
+function newClusterView(session) {
+  return session.run(GET_SERVERS)
+    .then((res) => {
+      session.close();
+      let record = res.records[0];
+      //Note we are loosing precision here but we are not
+      //terribly worried since it is only
+      //for dates more than 140000 years into the future.
+      let expires = record.get('ttl').toNumber();
+      let servers = record.get('servers');
+      let routers = new RoundRobinArray();
+      let readers = new RoundRobinArray();
+      let writers = new RoundRobinArray();
+      for (let i = 0; i <= servers.length; i++) {
+        let server = servers[i];
+
+        let role = server['role'];
+        let addresses = server['addresses'];
+        if (role === 'ROUTE') {
+          routers.pushAll(addresses);
+        } else if (role === 'WRITE') {
+          writers.pushAll(addresses);
+        } else if (role === 'READ') {
+          readers.pushAll(addresses);
+        }
+      }
+
+      return new ClusterView(expires, routers, readers, writers);
+    });
+}
+
 class RoutingDriver extends Driver {
 
   constructor(url, userAgent = 'neo4j-javascript/0.0', token = {}, config = {}) {
@@ -213,15 +254,25 @@ class RoutingDriver extends Driver {
     this._readers = new RoundRobinArray();
     this._writers = new RoundRobinArray();
     this._expires = Date.now();
-    this._checkServers();
   }
 
-  _checkServers() {
+  //TODO make nice, expose constants?
+  session(mode) {
+    //Check so that we have servers available
+    this._checkServers().then( () => {
+      let conn = this._acquireConnection(mode);
+      return this._createSession(conn);
+    });
+  }
+
+  async _checkServers() {
     if (this._expires < Date.now() ||
       this._routers.empty() ||
       this._readers.empty() ||
       this._writers.empty()) {
-      this._callServers();
+      return await this._callServers();
+    } else {
+      return new Promise((resolve, reject) => resolve(false));
     }
   }
 
@@ -239,16 +290,19 @@ class RoutingDriver extends Driver {
       let url = this._routers.hop();
       try {
         let res = await this._call(url);
+        console.log("got result");
         if (res.records.length != 1) continue;
         let record = res.records[0];
         //Note we are loosing precision here but we are not
         //terribly worried since it is only
         //for dates more than 140000 years into the future.
         this._expires += record.get('ttl').toNumber();
         let servers = record.get('servers');
+        console.log(servers);
         for (let i = 0; i <= servers.length; i++) {
           let server = servers[i];
-          seen.delete(server);
+          seen.remove(server);
+
           let role = server['role'];
           let addresses = server['addresses'];
           if (role === 'ROUTE') {
@@ -266,47 +320,42 @@ class RoutingDriver extends Driver {
         //these are no longer valid according to server
         let self = this;
         seen.forEach((key) => {
-          self._pool.purge(key);
+          console.log("remove seen");
+          self._pools.purge(key);
         });
         success = true;
-        return;
+        return new Promise((resolve, reject) => resolve(true));
       } catch (error) {
         //continue
-        this._forget(url);
         console.log(error);
+        this._forget(url);
       }
     }
 
+    let errorMsg = "Server could not perform discovery, please open a new driver with a different seed address.";
     if (this.onError) {
-      this.onError("Server could not perform discovery, please open a new driver with a different seed address.");
+      this.onError(errorMsg);
     }
-    this.close();
-  }
 
-  //TODO make nice, expose constants?
-  session(mode) {
-   let conn = this._aquireConnection(mode);
-    return this._createSession(conn);
+    return new Promise((resolve, reject) => reject(errorMsg));
   }
 
-  _aquireConnection(mode) {
+  _acquireConnection(mode) {
     //make sure we have enough servers
-    this._checkServers();
-
     let m = mode || WRITE;
     if (m === READ) {
-      return this._pools.acquire(this._readers.hop());
+      return this._pool.acquire(this._readers.hop());
     } else if (m === WRITE) {
-      return this._pools.acquire(this._writers.hop());
+      return this._pool.acquire(this._writers.hop());
     } else {
       //TODO fail
     }
   }
 
   _allServers() {
     let seen = new Set(this._routers.toArray());
-    let writers = this._writers.toArray()
-    let readers = this._readers.toArray()
+    let writers = this._writers.toArray();
+    let readers = this._readers.toArray();
     for (let i = 0; i < writers.length; i++) {
       seen.add(writers[i]);
     }
@@ -319,18 +368,19 @@ class RoutingDriver extends Driver {
   async _call(url) {
     let conn = this._pool.acquire(url);
     let session = this._createSession(conn);
-    console.log("calling " + GET_SERVERS);
     return session.run(GET_SERVERS)
       .then((res) => {
         session.close();
         return res;
       }).catch((err) => {
+        console.log(err);
         this._forget(url);
         return Promise.reject(err);
       });
   }
 
   _forget(url) {
+    console.log("forget");
     this._pools.purge(url);
     this._routers.remove(url);
     this._readers.remove(url);
@@ -426,9 +476,12 @@ let USER_AGENT = "neo4j-javascript/" + VERSION;
 function driver(url, authToken, config = {}) {
   let sch = scheme(url);
   if (sch === "bolt+routing://") {
-    return new RoutingDriver(url, USER_AGENT, authToken, config);
-  } else {
+      return new RoutingDriver(url, USER_AGENT, authToken, config);
+  } else if (sch === "bolt://") {
     return new Driver(url, USER_AGENT, authToken, config);
+  } else {
+    throw new Error("Unknown scheme: " + sch);
+
   }
 }
 

src/v1/session.js

@@ -33,19 +33,15 @@ import {newError} from "./error";
 class Session {
   /**
    * @constructor
-   * @param {Connection} conn - A connection to use
+   * @param {Promise} connectionPromise - A connection to use
    * @param {function()} onClose - Function to be called on connection close
    */
-  constructor( conn, onClose ) {
-    this._conn = conn;
+  constructor( connectionPromise, onClose ) {
+    this._connectionPromise = connectionPromise;
     this._onClose = onClose;
     this._hasTx = false;
   }
 
-  isEncrypted() {
-    return this._conn.isEncrypted();
-  }
-
   /**
    * Run Cypher statement
    * Could be called with a statement object i.e.: {statement: "MATCH ...", parameters: {param: 1}}
@@ -61,9 +57,11 @@ class Session {
     }
     let streamObserver = new _RunObserver();
     if (!this._hasTx) {
-      this._conn.run(statement, parameters, streamObserver);
-      this._conn.pullAll(streamObserver);
-      this._conn.sync();
+      this._connectionPromise.then((conn) => {
+        conn.run(statement, parameters, streamObserver);
+        conn.pullAll(streamObserver);
+        conn.sync();
+      }).catch((err) => streamObserver.onError(err));
     } else {
       streamObserver.onError(newError("Statements cannot be run directly on a "
        + "session with an open transaction; either run from within the "
@@ -82,13 +80,13 @@ class Session {
    */
   beginTransaction() {
     if (this._hasTx) {
-      throw new newError("You cannot begin a transaction on a session with an "
+      throw newError("You cannot begin a transaction on a session with an "
       + "open transaction; either run from within the transaction or use a "
       + "different session.")
     }
 
     this._hasTx = true;
-    return new Transaction(this._conn, () => {this._hasTx = false});
+    return new Transaction(this._connectionPromise, () => {this._hasTx = false});
   }
 
   /**

src/v1/transaction.js

@@ -27,14 +27,17 @@ import Result from './result';
 class Transaction {
   /**
    * @constructor
-   * @param {Connection} conn - A connection to use
+   * @param {Promise} connectionPromise - A connection to use
    * @param {function()} onClose - Function to be called when transaction is committed or rolled back.
    */
-  constructor(conn, onClose) {
-    this._conn = conn;
+  constructor(connectionPromise, onClose) {
+    this._connectionPromise = connectionPromise;
     let streamObserver = new _TransactionStreamObserver(this);
-    this._conn.run("BEGIN", {}, streamObserver);
-    this._conn.discardAll(streamObserver);
+    this._connectionPromise.then((conn) => {
+      conn.run("BEGIN", {}, streamObserver);
+      conn.discardAll(streamObserver);
+    }).catch(streamObserver.onError);
+
     this._state = _states.ACTIVE;
     this._onClose = onClose;
   }
@@ -52,7 +55,7 @@ class Transaction {
       parameters = statement.parameters || {};
       statement = statement.text;
     }
-    return this._state.run(this._conn,  new _TransactionStreamObserver(this), statement, parameters);
+    return this._state.run(this._connectionPromise,  new _TransactionStreamObserver(this), statement, parameters);
   }
 
   /**
@@ -63,7 +66,7 @@ class Transaction {
    * @returns {Result} - New Result
    */
   commit() {
-    let committed = this._state.commit(this._conn, new _TransactionStreamObserver(this));
+    let committed = this._state.commit(this._connectionPromise, new _TransactionStreamObserver(this));
     this._state = committed.state;
     //clean up
     this._onClose();
@@ -79,7 +82,7 @@ class Transaction {
    * @returns {Result} - New Result
    */
   rollback() {
-    let committed = this._state.rollback(this._conn, new _TransactionStreamObserver(this));
+    let committed = this._state.rollback(this._connectionPromise, new _TransactionStreamObserver(this));
     this._state = committed.state;
     //clean up
     this._onClose();
@@ -114,17 +117,20 @@ class _TransactionStreamObserver extends StreamObserver {
 let _states = {
   //The transaction is running with no explicit success or failure marked
   ACTIVE: {
-    commit: (conn, observer) => {
-      return {result: _runDiscardAll("COMMIT", conn, observer),
+    commit: (connectionPromise, observer) => {
+      return {result: _runDiscardAll("COMMIT", connectionPromise, observer),
         state: _states.SUCCEEDED}
     },
-    rollback: (conn, observer) => {
-      return {result: _runDiscardAll("ROLLBACK", conn, observer), state: _states.ROLLED_BACK};
+    rollback: (connectionPromise, observer) => {
+      return {result: _runDiscardAll("ROLLBACK", connectionPromise, observer), state: _states.ROLLED_BACK};
     },
-    run: (conn, observer, statement, parameters) => {
-      conn.run( statement, parameters || {}, observer );
-      conn.pullAll( observer );
-      conn.sync();
+    run: (connectionPromise, observer, statement, parameters) => {
+      connectionPromise.then((conn) => {
+        conn.run( statement, parameters || {}, observer );
+        conn.pullAll( observer );
+        conn.sync();
+      }).catch(observer.onError);
+
       return new Result( observer, statement, parameters );
     }
   },
@@ -196,10 +202,13 @@ let _states = {
   }
 };
 
-function _runDiscardAll(msg, conn, observer) {
-  conn.run(msg, {}, observer );
-  conn.discardAll(observer);
-  conn.sync();
+function _runDiscardAll(msg, connectionPromise, observer) {
+  connectionPromise.then((conn) => {
+    conn.run(msg, {}, observer);
+    conn.discardAll(observer);
+    conn.sync();
+  }).catch(observer.onError);
+
   return new Result(observer, msg, {});
 }
 

test/v1/boltkit.js

@@ -49,15 +49,15 @@ BoltKit.prototype.start = function(script, port) {
 
   var Server = function(){};
   //give process some time to exit
-  Server.prototype.exit = function(callback) {setTimeout(function(){callback(code);}, 500)};
+  Server.prototype.exit = function(callback) {setTimeout(function(){callback(code);}, 1000)};
 
   return new Server();
 };
 
 //Make sure boltstub is started before running
 //user code
 BoltKit.prototype.run = function(callback) {
-  setTimeout(callback, 500);
+  setTimeout(callback, 1000);
 };
 
 function boltKitSupport() {

test/v1/driver.test.js

@@ -48,6 +48,10 @@ describe('driver', function() {
     driver.session();
   });
 
+  it('should handle wrong scheme ', function() {
+    expect(function(){neo4j.driver("tank://localhost", neo4j.auth.basic("neo4j", "neo4j"))}).toThrow(new Error("Unknown scheme: tank://"));
+  });
+
   it('should fail early on wrong credentials', function(done) {
     // Given
     var driver = neo4j.driver("bolt://localhost", neo4j.auth.basic("neo4j", "who would use such a password"));