Routing support for running on session.

Author: pontusmelke

src/v1/driver.js

@@ -108,15 +108,12 @@ class Driver {
    * it is returned to the pool, the session will be reset to a clean state and
    * made available for others to use.
    *
+   * @param {String} mode of session - optional
    * @return {Session} new session.
    */
-  session() {
-    let conn = this._pool.acquire(this._url);
-    return this._createSession(Promise.resolve(conn));
-  }
-
-  _createSession(connectionPromise) {
-    return new Session(connectionPromise, (cb) => {
+  session(mode) {
+    let connectionPromise = this._acquireConnection(mode);
+    return this._createSession(connectionPromise, (cb) => {
       // This gets called on Session#close(), and is where we return
       // the pooled 'connection' instance.
 
@@ -134,14 +131,23 @@ class Driver {
         conn._release();
       });
 
-
       // Call user callback
       if (cb) {
         cb();
       }
     });
   }
 
+  //Extension point
+  _acquireConnection(mode) {
+   return Promise.resolve(this._pool.acquire(this._url));
+  }
+
+  //Extension point
+  _createSession(connectionPromise, cb) {
+    return new Session(connectionPromise, cb);
+  }
+
   /**
    * Close all open sessions and other associated resources. You should
    * make sure to use this when you are done with this driver instance.
@@ -152,6 +158,8 @@ class Driver {
       if (this._openSessions.hasOwnProperty(sessionId)) {
         this._openSessions[sessionId].close();
       }
+
+      this._pool.purgeAll();
     }
   }
 }

src/v1/internal/ch-node.js

@@ -293,9 +293,10 @@ class NodeChannel {
   }
 
   _handleConnectionError( err ) {
-    this._error = err;
+    let msg = err.message || 'Failed to connect to server';
+    this._error = newError(msg, SESSION_EXPIRED);
     if( this.onerror ) {
-      this.onerror(err);
+      this.onerror(this._error);
     }
   }
 

src/v1/internal/pool.js

@@ -41,7 +41,11 @@ class Pool {
 
   acquire(key) {
     let resource;
-    let pool = this._pools[key] || [];
+    let pool = this._pools[key];
+    if (!pool) {
+      pool = [];
+      this._pools[key] = pool;
+    }
     while (pool.length) {
       resource = pool.pop();
 
@@ -65,15 +69,23 @@ class Pool {
     delete this._pools[key]
   }
 
+  purgeAll() {
+    for (let key in this._pools.keys) {
+      if (this._pools.hasOwnPropertykey) {
+        this.purge(key);
+      }
+    }
+  }
+
   has(key) {
     return (key in this._pools);
   }
 
   _release(key, resource) {
     let pool = this._pools[key];
     if (!pool) {
-      pool = [];
-      this._pools[key] = pool;
+      //key has been purged, don't put it back
+      return;
     }
     if( pool.length >= this._maxIdle || !this._validate(resource) ) {
       this._destroy(resource);

src/v1/internal/stream-observer.js

@@ -32,14 +32,16 @@ import Record from "../record";
 class StreamObserver {
   /**
    * @constructor
+   * @param errorCallback optional callback to be used for adding additional logic on error
    */
-  constructor() {
+  constructor(errorCallback = () => {}) {
     this._fieldKeys = null;
     this._fieldLookup = null;
     this._queuedRecords = [];
     this._tail = null;
     this._error = null;
     this._hasFailed = false;
+    this._errorCallback = errorCallback;
   }
 
   /**
@@ -81,25 +83,30 @@ class StreamObserver {
     }
   }
 
+  resolveConnection(conn) {
+    this._conn = conn;
+  }
+
   /**
    * Will be called on errors.
    * If user-provided observer is present, pass the error
    * to it's onError method, otherwise set instance variable _error.
    * @param {Object} error - An error object
    */
   onError(error) {
+    let transformedError = this._errorCallback(error, this._conn);
     if(this._hasFailed) {
       return;
     }
     this._hasFailed = true;
     if( this._observer ) {
       if( this._observer.onError ) {
-        this._observer.onError( error );
+        this._observer.onError( transformedError );
       } else {
-        console.log( error );
+        console.log( transformedError );
       }
     } else {
-      this._error = error;
+      this._error = transformedError;
     }
   }
 

src/v1/routing-driver.js

@@ -23,8 +23,6 @@ import {newError, SERVICE_UNAVAILABLE, SESSION_EXPIRED} from "./error";
 import RoundRobinArray from './internal/round-robin-array';
 import "babel-polyfill";
 
-
-
 /**
  * A driver that supports routing in a core-edge cluster.
  */
@@ -35,9 +33,40 @@ class RoutingDriver extends Driver {
     this._clusterView = new ClusterView(new RoundRobinArray([url]));
   }
 
-  session(mode) {
-    let conn = this._acquireConnection(mode);
-    return this._createSession(conn);
+  _createSession(connectionPromise, cb) {
+    return new RoutingSession(connectionPromise, cb, (err, conn) => {
+      let code = err.code;
+      if (!code) {
+        try {
+          code = err.fields[0].code;
+        } catch (e) {
+          code = 'UNKNOWN';
+        }
+      }
+
+      if (code === SERVICE_UNAVAILABLE || code === SESSION_EXPIRED) {
+        if (conn) {
+          this._forget(conn.url)
+        } else {
+          connectionPromise.then((conn) => {
+            this._forget(conn.url);
+          });
+        }
+        return err;
+      } else if (code === 'Neo.ClientError.Cluster.NotALeader') {
+        let url = 'UNKNOWN';
+        if (conn) {
+          url = conn.url;
+          this._clusterView.writers.remove(conn.url);
+        } else {
+          connectionPromise.then((conn) => {
+            this._clusterView.writers.remove(conn.url);
+          });
+        }
+
+        return newError("No longer possible to write to server at " + url, SESSION_EXPIRED);
+      }
+    });
   }
 
   _updatedClusterView() {
@@ -60,6 +89,7 @@ class RoutingDriver extends Driver {
       return acc;
     }
   }
+
   _diff(oldView, updatedView) {
     let oldSet = oldView.all();
     let newSet = updatedView.all();
@@ -81,9 +111,17 @@ class RoutingDriver extends Driver {
       //update our cached view
       this._clusterView = view;
       if (m === READ) {
-        return this._pool.acquire(view.readers.hop());
+        let key = view.readers.hop();
+        if (!key) {
+          return Promise.reject(newError('No read servers available', SESSION_EXPIRED));
+        }
+        return this._pool.acquire(key);
       } else if (m === WRITE) {
-        return this._pool.acquire(view.writers.hop());
+        let key = view.writers.hop();
+        if (!key) {
+          return Promise.reject(newError('No write servers available', SESSION_EXPIRED));
+        }
+        return this._pool.acquire(key);
       } else {
         return Promise.reject(m + " is not a valid option");
       }
@@ -96,7 +134,6 @@ class RoutingDriver extends Driver {
   }
 }
 
-
 class ClusterView {
   constructor(routers, readers, writers, expires) {
     this.routers = routers || new RoundRobinArray();
@@ -133,6 +170,17 @@ class ClusterView {
   }
 }
 
+class RoutingSession extends Session {
+  constructor(connectionPromise, onClose, onFailedConnection) {
+    super(connectionPromise, onClose);
+    this._onFailedConnection = onFailedConnection;
+  }
+
+  _onRunFailure() {
+    return this._onFailedConnection;
+  }
+}
+
 let GET_SERVERS = "CALL dbms.cluster.routing.getServers";
 
 /**
@@ -170,6 +218,9 @@ function newClusterView(session) {
         }
       }
       return new ClusterView(routers, readers, writers, expires);
+    })
+    .catch(() => {
+      return Promise.reject(newError("No servers could be found at this instant.", SERVICE_UNAVAILABLE));
     });
 }
 

src/v1/session.js

@@ -55,9 +55,10 @@ class Session {
       parameters = statement.parameters || {};
       statement = statement.text;
     }
-    let streamObserver = new _RunObserver();
+    let streamObserver = new _RunObserver(this._onRunFailure());
     if (!this._hasTx) {
       this._connectionPromise.then((conn) => {
+        streamObserver.resolveConnection(conn);
         conn.run(statement, parameters, streamObserver);
         conn.pullAll(streamObserver);
         conn.sync();
@@ -105,12 +106,17 @@ class Session {
       cb();
     }
   }
+
+  //Can be overridden to add error callback on RUN
+  _onRunFailure() {
+    return (err) => {return err};
+  }
 }
 
 /** Internal stream observer used for transactional results*/
 class _RunObserver extends StreamObserver {
-  constructor() {
-    super();
+  constructor(onError) {
+    super(onError);
     this._meta = {};
   }
 

test/resources/boltkit/non_discovery.script

@@ -0,0 +1,8 @@
+!: AUTO INIT
+!: AUTO RESET
+!: AUTO PULL_ALL
+
+C: RUN "CALL dbms.cluster.routing.getServers" {}
+C: PULL_ALL
+S: FAILURE {"code": "Neo.ClientError.Procedure.ProcedureNotFound", "message": "blabla"}
+S: IGNORED

test/resources/boltkit/not_able_to_write.script

@@ -0,0 +1,11 @@
+!: AUTO INIT
+!: AUTO RESET
+!: AUTO PULL_ALL
+!: AUTO RUN "ROLLBACK" {}
+!: AUTO RUN "BEGIN" {}
+!: AUTO PULL_ALL
+
+C: RUN "CREATE ()" {}
+C: PULL_ALL
+S: FAILURE {"code": "Neo.ClientError.Cluster.NotALeader", "message": "blabla"}
+S: IGNORED

test/resources/boltkit/rediscover.script

@@ -0,0 +1,14 @@
+!: AUTO INIT
+!: AUTO RESET
+!: AUTO PULL_ALL
+
+C: RUN "CALL dbms.cluster.routing.getServers" {}
+   PULL_ALL
+S: SUCCESS {"fields": ["ttl", "servers"]}
+   RECORD [9223372036854775807, [{"addresses": ["127.0.0.1:9001"], "role": "ROUTE"}]]
+   SUCCESS {}
+C: RUN "CALL dbms.cluster.routing.getServers" {}
+   PULL_ALL
+S: SUCCESS {"fields": ["ttl", "servers"]}
+   RECORD [9223372036854775807, [{"addresses": ["127.0.0.1:9004"],"role": "WRITE"}, {"addresses": ["127.0.0.1:9005"], "role": "READ"},{"addresses": ["127.0.0.1:9002","127.0.0.1:9003","127.0.0.1:9004"], "role": "ROUTE"}]]
+   SUCCESS {}