Support for routing when running on transactions

Author: pontusmelke

src/v1/internal/stream-observer.js

@@ -34,7 +34,7 @@ class StreamObserver {
    * @constructor
    * @param errorCallback optional callback to be used for adding additional logic on error
    */
-  constructor(errorCallback = () => {}) {
+  constructor(errorCallback = (err) => {return err}) {
     this._fieldKeys = null;
     this._fieldLookup = null;
     this._queuedRecords = [];

src/v1/session.js

@@ -87,7 +87,7 @@ class Session {
     }
 
     this._hasTx = true;
-    return new Transaction(this._connectionPromise, () => {this._hasTx = false});
+    return new Transaction(this._connectionPromise, () => {this._hasTx = false}, this._onRunFailure());
   }
 
   /**

src/v1/transaction.js

@@ -30,16 +30,18 @@ class Transaction {
    * @param {Promise} connectionPromise - A connection to use
    * @param {function()} onClose - Function to be called when transaction is committed or rolled back.
    */
-  constructor(connectionPromise, onClose) {
+  constructor(connectionPromise, onClose, errorTransformer) {
     this._connectionPromise = connectionPromise;
     let streamObserver = new _TransactionStreamObserver(this);
     this._connectionPromise.then((conn) => {
+      streamObserver.resolveConnection(conn);
       conn.run("BEGIN", {}, streamObserver);
       conn.discardAll(streamObserver);
     }).catch(streamObserver.onError);
 
     this._state = _states.ACTIVE;
     this._onClose = onClose;
+    this._errorTransformer = errorTransformer;
   }
 
   /**
@@ -98,7 +100,7 @@ class Transaction {
 /** Internal stream observer used for transactional results*/
 class _TransactionStreamObserver extends StreamObserver {
   constructor(tx) {
-    super();
+    super(tx._errorTransformer || ((err) => {return err}));
     this._tx = tx;
     //this is to to avoid multiple calls to onError caused by IGNORED
     this._hasFailed = false;
@@ -126,6 +128,7 @@ let _states = {
     },
     run: (connectionPromise, observer, statement, parameters) => {
       connectionPromise.then((conn) => {
+        observer.resolveConnection(conn);
         conn.run( statement, parameters || {}, observer );
         conn.pullAll( observer );
         conn.sync();
@@ -204,6 +207,7 @@ let _states = {
 
 function _runDiscardAll(msg, connectionPromise, observer) {
   connectionPromise.then((conn) => {
+    observer.resolveConnection(conn);
     conn.run(msg, {}, observer);
     conn.discardAll(observer);
     conn.sync();

test/resources/boltkit/not_able_to_write.script

@@ -1,6 +1,7 @@
 !: AUTO INIT
 !: AUTO RESET
 !: AUTO PULL_ALL
+!: AUTO DISCARD_ALL
 !: AUTO RUN "ROLLBACK" {}
 !: AUTO RUN "BEGIN" {}
 !: AUTO PULL_ALL

test/resources/boltkit/not_able_to_write_in_transaction.script

@@ -0,0 +1,14 @@
+!: AUTO INIT
+!: AUTO RESET
+!: AUTO PULL_ALL
+!: AUTO DISCARD_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
+C: RUN "COMMIT" {}
+S: IGNORED

test/v1/routing.driver.test.js

@@ -487,5 +487,39 @@ describe('routing driver ', function() {
       });
     });
   });
+
+  it('should handle leader switch while writing on transaction', function (done) {
+    if (!boltkit.BoltKitSupport) {
+      done();
+      return;
+    }
+    // Given
+    var kit = new boltkit.BoltKit(true);
+    var seedServer = kit.start('./test/resources/boltkit/acquire_endpoints.script', 9001);
+    var readServer = kit.start('./test/resources/boltkit/not_able_to_write_in_transaction.script', 9007);
+
+    kit.run(function () {
+      var driver = neo4j.driver("bolt+routing://127.0.0.1:9001", neo4j.auth.basic("neo4j", "neo4j"));
+      // When
+      var session = driver.session();
+      var tx = session.beginTransaction();
+      tx.run("CREATE ()");
+
+      tx.commit().catch(function (err) {
+        //the server at 9007 should have been removed
+        expect(driver._clusterView.writers.toArray()).toEqual([ '127.0.0.1:9008']);
+        expect(err.code).toEqual(neo4j.SESSION_EXPIRED);
+        session.close();
+        driver.close();
+        seedServer.exit(function (code1) {
+          readServer.exit(function (code2) {
+            expect(code1).toEqual(0);
+            expect(code2).toEqual(0);
+            done();
+          });
+        });
+      });
+    });
+  });
 });