Expose transaction operations on BoltProtocol

Make it expose logical operations for explicit transactions like begin,
commit and rollback. These operations require different set of messages
in Bolt V3. Having operations exposed like this should make it easier
to override them.

Author: lutovich

src/v1/internal/bolt-protocol.js renamed to src/v1/internal/bolt-protocol-v1.js

@@ -59,6 +59,35 @@ export default class BoltProtocol {
     this._connection.write(message, observer, true);
   }
 
+  /**
+   * Begin an explicit transaction.
+   * @param {Bookmark} bookmark the bookmark.
+   * @param {StreamObserver} observer the response observer.
+   */
+  beginTransaction(bookmark, observer) {
+    const runMessage = RequestMessage.run('BEGIN', bookmark.asBeginTransactionParameters());
+    const pullAllMessage = RequestMessage.pullAll();
+
+    this._connection.write(runMessage, observer, false);
+    this._connection.write(pullAllMessage, observer, false);
+  }
+
+  /**
+   * Commit the explicit transaction.
+   * @param {StreamObserver} observer the response observer.
+   */
+  commitTransaction(observer) {
+    this.run('COMMIT', {}, observer);
+  }
+
+  /**
+   * Rollback the explicit transaction.
+   * @param {StreamObserver} observer the response observer.
+   */
+  rollbackTransaction(observer) {
+    this.run('ROLLBACK', {}, observer);
+  }
+
   /**
    * Send a Cypher statement through the underlying connection.
    * @param {string} statement the cypher statement.

src/v1/internal/protocol-handshaker.js

@@ -21,7 +21,7 @@ import {alloc} from './buf';
 import {newError} from '../error';
 import * as v1 from './packstream-v1';
 import * as v2 from './packstream-v2';
-import BoltProtocol from './bolt-protocol';
+import BoltProtocol from './bolt-protocol-v1';
 
 const HTTP_MAGIC_PREAMBLE = 1213486160; // == 0x48545450 == "HTTP"
 const BOLT_MAGIC_PREAMBLE = 0x6060B017;

src/v1/transaction.js

@@ -41,7 +41,7 @@ class Transaction {
     const streamObserver = new _TransactionStreamObserver(this);
 
     this._connectionHolder.getConnection(streamObserver)
-      .then(conn => conn.protocol().run('BEGIN', bookmark.asBeginTransactionParameters(), streamObserver))
+      .then(conn => conn.protocol().beginTransaction(bookmark, streamObserver))
       .catch(error => streamObserver.onError(error));
 
     this._state = _states.ACTIVE;
@@ -151,11 +151,16 @@ let _states = {
   //The transaction is running with no explicit success or failure marked
   ACTIVE: {
     commit: (connectionHolder, observer) => {
-      return {result: _runPullAll("COMMIT", connectionHolder, observer),
-        state: _states.SUCCEEDED}
+      return {
+        result: finishTransaction(true, connectionHolder, observer),
+        state: _states.SUCCEEDED
+      };
     },
     rollback: (connectionHolder, observer) => {
-      return {result: _runPullAll("ROLLBACK", connectionHolder, observer), state: _states.ROLLED_BACK};
+      return {
+        result: finishTransaction(false, connectionHolder, observer),
+        state: _states.ROLLED_BACK
+      };
     },
     run: (connectionHolder, observer, statement, parameters) => {
       connectionHolder.getConnection(observer)
@@ -233,14 +238,20 @@ let _states = {
   }
 };
 
-function _runPullAll(msg, connectionHolder, observer) {
+function finishTransaction(commit, connectionHolder, observer) {
   connectionHolder.getConnection(observer)
-    .then(conn => conn.protocol().run(msg, {}, observer))
+    .then(connection => {
+      if (commit) {
+        return connection.protocol().commitTransaction(observer);
+      } else {
+        return connection.protocol().rollbackTransaction(observer);
+      }
+    })
     .catch(error => observer.onError(error));
 
   // for commit & rollback we need result that uses real connection holder and notifies it when
   // connection is not needed and can be safely released to the pool
-  return new Result(observer, msg, {}, emptyMetadataSupplier, connectionHolder);
+  return new Result(observer, '', {}, emptyMetadataSupplier, connectionHolder);
 }
 
 /**

test/internal/bolt-protocol.test.js renamed to test/internal/bolt-protocol-v1.test.js

@@ -17,8 +17,9 @@
  * limitations under the License.
  */
 
-import BoltProtocol from '../../src/v1/internal/bolt-protocol';
+import BoltProtocol from '../../src/v1/internal/bolt-protocol-v1';
 import RequestMessage from '../../src/v1/internal/request-message';
+import Bookmark from '../../src/v1/internal/bookmark';
 
 class MessageRecorder {
 
@@ -55,8 +56,8 @@ describe('BoltProtocol', () => {
 
     recorder.verifyMessageCount(1);
     verifyMessage(RequestMessage.init(clientName, authToken), recorder.messages[0]);
-    expect(recorder.observers[0]).toBe(observer);
-    expect(recorder.flushes[0]).toBeTruthy();
+    expect(recorder.observers).toEqual([observer]);
+    expect(recorder.flushes).toEqual([true]);
   });
 
   it('should run a statement', () => {
@@ -74,11 +75,8 @@ describe('BoltProtocol', () => {
     verifyMessage(RequestMessage.run(statement, parameters), recorder.messages[0]);
     verifyMessage(RequestMessage.pullAll(), recorder.messages[1]);
 
-    expect(recorder.observers[0]).toBe(observer);
-    expect(recorder.observers[1]).toBe(observer);
-
-    expect(recorder.flushes[0]).toBeFalsy();
-    expect(recorder.flushes[1]).toBeTruthy();
+    expect(recorder.observers).toEqual([observer, observer]);
+    expect(recorder.flushes).toEqual([false, true]);
   });
 
   it('should reset the connection', () => {
@@ -91,8 +89,60 @@ describe('BoltProtocol', () => {
 
     recorder.verifyMessageCount(1);
     verifyMessage(RequestMessage.reset(), recorder.messages[0]);
-    expect(recorder.observers[0]).toBe(observer);
-    expect(recorder.flushes[0]).toBeTruthy();
+    expect(recorder.observers).toEqual([observer]);
+    expect(recorder.flushes).toEqual([true]);
+  });
+
+  it('should begin a transaction', () => {
+    const recorder = new MessageRecorder();
+    const protocol = new BoltProtocol(recorder, null, null);
+
+    const bookmark = new Bookmark('neo4j:bookmark:v1:tx42');
+    const observer = {};
+
+    protocol.beginTransaction(bookmark, observer);
+
+    recorder.verifyMessageCount(2);
+
+    verifyMessage(RequestMessage.run('BEGIN', bookmark.asBeginTransactionParameters()), recorder.messages[0]);
+    verifyMessage(RequestMessage.pullAll(), recorder.messages[1]);
+
+    expect(recorder.observers).toEqual([observer, observer]);
+    expect(recorder.flushes).toEqual([false, false]);
+  });
+
+  it('should commit a transaction', () => {
+    const recorder = new MessageRecorder();
+    const protocol = new BoltProtocol(recorder, null, null);
+
+    const observer = {};
+
+    protocol.commitTransaction(observer);
+
+    recorder.verifyMessageCount(2);
+
+    verifyMessage(RequestMessage.run('COMMIT', {}), recorder.messages[0]);
+    verifyMessage(RequestMessage.pullAll(), recorder.messages[1]);
+
+    expect(recorder.observers).toEqual([observer, observer]);
+    expect(recorder.flushes).toEqual([false, true]);
+  });
+
+  it('should rollback a transaction', () => {
+    const recorder = new MessageRecorder();
+    const protocol = new BoltProtocol(recorder, null, null);
+
+    const observer = {};
+
+    protocol.rollbackTransaction(observer);
+
+    recorder.verifyMessageCount(2);
+
+    verifyMessage(RequestMessage.run('ROLLBACK', {}), recorder.messages[0]);
+    verifyMessage(RequestMessage.pullAll(), recorder.messages[1]);
+
+    expect(recorder.observers).toEqual([observer, observer]);
+    expect(recorder.flushes).toEqual([false, true]);
   });
 
 });

test/internal/protocol-handshaker.test.js

@@ -19,7 +19,7 @@
 
 import ProtocolHandshaker from '../../src/v1/internal/protocol-handshaker';
 import Logger from '../../src/v1/internal/logger';
-import BoltProtocol from '../../src/v1/internal/bolt-protocol';
+import BoltProtocol from '../../src/v1/internal/bolt-protocol-v1';
 import {alloc} from '../../src/v1/internal/buf';
 
 describe('ProtocolHandshaker', () => {