Merge pull request #216 from lutovich/1.2-initial-bookmark-in-session

Allow initial bookmark in session

Author: Zhen Li

src/v1/driver.js

@@ -108,12 +108,14 @@ 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
+   * @param {string} [mode=WRITE] the access mode of this session, allowed values are {@link READ} and {@link WRITE}.
+   * @param {string} [bookmark=null] the initial reference to some previous transaction. Value is optional and
+   * absence indicates that that the bookmark does not exist or is unknown.
    * @return {Session} new session.
    */
-  session(mode) {
+  session(mode, bookmark) {
     const sessionMode = Driver._validateSessionMode(mode);
-    return this._createSession(sessionMode, this._connectionProvider);
+    return this._createSession(sessionMode, this._connectionProvider, bookmark);
   }
 
   static _validateSessionMode(rawMode) {
@@ -130,8 +132,8 @@ class Driver {
   }
 
   //Extension point
-  _createSession(mode, connectionProvider) {
-    return new Session(mode, connectionProvider);
+  _createSession(mode, connectionProvider, bookmark) {
+    return new Session(mode, connectionProvider, bookmark);
   }
 
   _driverOnErrorCallback(error) {

src/v1/routing-driver.js

@@ -35,8 +35,8 @@ class RoutingDriver extends Driver {
     return new LoadBalancer(address, connectionPool, driverOnErrorCallback);
   }
 
-  _createSession(mode, connectionProvider) {
-    return new RoutingSession(mode, connectionProvider, (error, conn) => {
+  _createSession(mode, connectionProvider, bookmark) {
+    return new RoutingSession(mode, connectionProvider, bookmark, (error, conn) => {
       if (error.code === SERVICE_UNAVAILABLE || error.code === SESSION_EXPIRED) {
         // connection is undefined if error happened before connection was acquired
         if (conn) {
@@ -66,8 +66,8 @@ class RoutingDriver extends Driver {
 }
 
 class RoutingSession extends Session {
-  constructor(mode, connectionProvider, onFailedConnection) {
-    super(mode, connectionProvider);
+  constructor(mode, connectionProvider, bookmark, onFailedConnection) {
+    super(mode, connectionProvider, bookmark);
     this._onFailedConnection = onFailedConnection;
   }
 

src/v1/session.js

@@ -34,15 +34,17 @@ class Session {
 
   /**
    * @constructor
-   * @param {string} mode - the default access mode for this session.
-   * @param connectionProvider - the connection provider to acquire connections from.
+   * @param {string} mode the default access mode for this session.
+   * @param {ConnectionProvider} connectionProvider - the connection provider to acquire connections from.
+   * @param {string} bookmark - the initial bookmark for this session.
    */
-  constructor(mode, connectionProvider) {
+  constructor(mode, connectionProvider, bookmark) {
     this._mode = mode;
     this._readConnectionHolder = new ConnectionHolder(READ, connectionProvider);
     this._writeConnectionHolder = new ConnectionHolder(WRITE, connectionProvider);
     this._open = true;
     this._hasTx = false;
+    this._lastBookmark = bookmark;
   }
 
   /**
@@ -84,11 +86,17 @@ class Session {
    *
    * While a transaction is open the session cannot be used to run statements outside the transaction.
    *
+   * @param {string} bookmark - a reference to a previous transaction. DEPRECATED: This function is deprecated in
+   * favour of {@link Driver#session(string)} that accepts an initial bookmark. Session will ensure that all nested
+   * transactions are chained with bookmarks to guarantee causal consistency.
    * @returns {Transaction} - New Transaction
    */
   beginTransaction(bookmark) {
     if (bookmark) {
-      assertString(bookmark, "Bookmark");
+      assertString(bookmark, 'Bookmark');
+    }
+    if (typeof bookmark !== 'undefined') {
+      this._lastBookmark = bookmark;
     }
 
     if (this._hasTx) {
@@ -99,21 +107,25 @@ class Session {
 
     this._hasTx = true;
 
-    // todo: add mode parameter
-    const mode = this._mode;
-    const connectionHolder = this._connectionHolderWithMode(mode);
+    const connectionHolder = this._connectionHolderWithMode(this._mode);
     connectionHolder.initializeConnection();
 
     return new Transaction(connectionHolder, () => {
         this._hasTx = false;
       },
-      this._onRunFailure(), bookmark, (bookmark) => {this._lastBookmark = bookmark});
+      this._onRunFailure(), this._lastBookmark, this._updateBookmark.bind(this));
   }
 
   lastBookmark() {
     return this._lastBookmark;
   }
 
+  _updateBookmark(newBookmark) {
+    if (newBookmark) {
+      this._lastBookmark = newBookmark;
+    }
+  }
+
   /**
    * Close this session.
    * @param {function()} callback - Function to be called after the session has been closed

test/resources/boltkit/acquire_endpoints_with_one_of_each.script

@@ -0,0 +1,9 @@
+!: 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:9007"],"role": "WRITE"}, {"addresses": ["127.0.0.1:9005"], "role": "READ"},{"addresses": ["127.0.0.1:9001","127.0.0.1:9002"], "role": "ROUTE"}]]
+   SUCCESS {}

test/resources/boltkit/read_tx_with_bookmarks.script

@@ -0,0 +1,18 @@
+!: AUTO INIT
+!: AUTO RESET
+!: AUTO PULL_ALL
+
+C: RUN "BEGIN" {"bookmark": "OldBookmark"}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {}
+C: RUN "MATCH (n) RETURN n.name AS name" {}
+   PULL_ALL
+S: SUCCESS {"fields": ["name"]}
+   RECORD ["Bob"]
+   RECORD ["Alice"]
+   SUCCESS {"bookmark": "NewBookmark"}
+C: RUN "COMMIT" {}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {}

test/resources/boltkit/write_read_tx_with_bookmark_override.script.mst

@@ -0,0 +1,30 @@
+!: AUTO INIT
+!: AUTO RESET
+!: AUTO PULL_ALL
+
+C: RUN "BEGIN" {"bookmark": "BookmarkA"}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {}
+C: RUN "CREATE (n {name:'Bob'})" {}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {"bookmark": "BookmarkB"}
+C: RUN "COMMIT" {}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {}
+C: RUN "BEGIN" {{{bookmarkOverride}}}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {}
+C: RUN "MATCH (n) RETURN n.name AS name" {}
+   PULL_ALL
+S: SUCCESS {"fields": ["name"]}
+   RECORD ["Bob"]
+   SUCCESS {"bookmark": "BookmarkC"}
+C: RUN "COMMIT" {}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {}
+

test/resources/boltkit/write_read_tx_with_bookmarks.script

@@ -0,0 +1,30 @@
+!: AUTO INIT
+!: AUTO RESET
+!: AUTO PULL_ALL
+
+C: RUN "BEGIN" {"bookmark": "BookmarkA"}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {}
+C: RUN "CREATE (n {name:'Bob'})" {}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {"bookmark": "BookmarkB"}
+C: RUN "COMMIT" {}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {}
+C: RUN "BEGIN" {"bookmark": "BookmarkB"}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {}
+C: RUN "MATCH (n) RETURN n.name AS name" {}
+   PULL_ALL
+S: SUCCESS {"fields": ["name"]}
+   RECORD ["Bob"]
+   SUCCESS {"bookmark": "BookmarkC"}
+C: RUN "COMMIT" {}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {}
+

test/resources/boltkit/write_tx_with_bookmarks.script

@@ -0,0 +1,16 @@
+!: AUTO INIT
+!: AUTO RESET
+!: AUTO PULL_ALL
+
+C: RUN "BEGIN" {"bookmark": "OldBookmark"}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {}
+C: RUN "CREATE (n {name:'Bob'})" {}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {"bookmark": "NewBookmark"}
+C: RUN "COMMIT" {}
+   PULL_ALL
+S: SUCCESS {}
+   SUCCESS {}