Merge pull request #444 from ali-ince/1.7-pass-access-mode

Pass AccessMode in BEGIN and RUN messages

Author: PetrJanouch

src/v1/driver.js

@@ -28,6 +28,7 @@ import ConnectivityVerifier from './internal/connectivity-verifier';
 import PoolConfig, {DEFAULT_ACQUISITION_TIMEOUT, DEFAULT_MAX_SIZE} from './internal/pool-config';
 import Logger from './internal/logger';
 import ConnectionErrorHandler from './internal/connection-error-handler';
+import {ACCESS_MODE_READ, ACCESS_MODE_WRITE} from './internal/constants';
 
 const DEFAULT_MAX_CONNECTION_LIFETIME = 60 * 60 * 1000; // 1 hour
 
@@ -36,14 +37,14 @@ const DEFAULT_MAX_CONNECTION_LIFETIME = 60 * 60 * 1000; // 1 hour
  * Should be used like this: `driver.session(neo4j.session.READ)`.
  * @type {string}
  */
-const READ = 'READ';
+const READ = ACCESS_MODE_READ;
 
 /**
  * Constant that represents write session access mode.
  * Should be used like this: `driver.session(neo4j.session.WRITE)`.
  * @type {string}
  */
-const WRITE = 'WRITE';
+const WRITE = ACCESS_MODE_WRITE;
 
 let idGenerator = 0;
 
@@ -174,15 +175,15 @@ class Driver {
   }
 
   /**
-   * Acquire a session to communicate with the database. The driver maintains
-   * a pool of sessions, so calling this method is normally cheap because you
-   * will be pulling a session out of the common pool.
+   * Acquire a session to communicate with the database. The session will
+   * borrow connections from the underlying connection pool as required and
+   * should be considered lightweight and disposable.
    *
    * This comes with some responsibility - make sure you always call
    * {@link close} when you are done using a session, and likewise,
    * make sure you don't close your session before you are done using it. Once
-   * it is returned to the pool, the session will be reset to a clean state and
-   * made available for others to use.
+   * it is closed, the underlying connection will be released to the connection
+   * pool and made available for others to use.
    *
    * @param {string} [mode=WRITE] the access mode of this session, allowed values are {@link READ} and {@link WRITE}.
    * @param {string|string[]} [bookmarkOrBookmarks=null] the initial reference or references to some previous
@@ -198,7 +199,7 @@ class Driver {
 
   static _validateSessionMode(rawMode) {
     const mode = rawMode || WRITE;
-    if (mode !== READ && mode !== WRITE) {
+    if (mode !== ACCESS_MODE_READ && mode !== ACCESS_MODE_WRITE) {
       throw newError('Illegal session mode ' + mode);
     }
     return mode;

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

@@ -21,6 +21,7 @@ import * as v1 from './packstream-v1';
 import {newError} from '../error';
 import Bookmark from './bookmark';
 import TxConfig from './tx-config';
+import {ACCESS_MODE_WRITE} from "./constants";
 
 export default class BoltProtocol {
 
@@ -80,9 +81,10 @@ export default class BoltProtocol {
    * Begin an explicit transaction.
    * @param {Bookmark} bookmark the bookmark.
    * @param {TxConfig} txConfig the configuration.
+   * @param {string} mode the access mode.
    * @param {StreamObserver} observer the response observer.
    */
-  beginTransaction(bookmark, txConfig, observer) {
+  beginTransaction(bookmark, txConfig, mode, observer) {
     assertTxConfigIsEmpty(txConfig, this._connection, observer);
 
     const runMessage = RequestMessage.run('BEGIN', bookmark.asBeginTransactionParameters());
@@ -97,15 +99,19 @@ export default class BoltProtocol {
    * @param {StreamObserver} observer the response observer.
    */
   commitTransaction(observer) {
-    this.run('COMMIT', {}, Bookmark.empty(), TxConfig.empty(), observer);
+    // WRITE access mode is used as a place holder here, it has
+    // no effect on behaviour for Bolt V1 & V2
+    this.run('COMMIT', {}, Bookmark.empty(), TxConfig.empty(), ACCESS_MODE_WRITE, observer);
   }
 
   /**
    * Rollback the explicit transaction.
    * @param {StreamObserver} observer the response observer.
    */
   rollbackTransaction(observer) {
-    this.run('ROLLBACK', {}, Bookmark.empty(), TxConfig.empty(), observer);
+    // WRITE access mode is used as a place holder here, it has
+    // no effect on behaviour for Bolt V1 & V2
+    this.run('ROLLBACK', {}, Bookmark.empty(), TxConfig.empty(), ACCESS_MODE_WRITE, observer);
   }
 
   /**
@@ -114,10 +120,11 @@ export default class BoltProtocol {
    * @param {object} parameters the statement parameters.
    * @param {Bookmark} bookmark the bookmark.
    * @param {TxConfig} txConfig the auto-commit transaction configuration.
+   * @param {string} mode the access mode.
    * @param {StreamObserver} observer the response observer.
    */
-  run(statement, parameters, bookmark, txConfig, observer) {
-    // bookmark is ignored in this version of the protocol
+  run(statement, parameters, bookmark, txConfig, mode, observer) {
+    // bookmark and mode are ignored in this versioon of the protocol
     assertTxConfigIsEmpty(txConfig, this._connection, observer);
 
     const runMessage = RequestMessage.run(statement, parameters);

src/v1/internal/bolt-protocol-v3.js

@@ -52,9 +52,9 @@ export default class BoltProtocol extends BoltProtocolV2 {
     this._connection.write(message, observer, true);
   }
 
-  beginTransaction(bookmark, txConfig, observer) {
+  beginTransaction(bookmark, txConfig, mode, observer) {
     prepareToHandleSingleResponse(observer);
-    const message = RequestMessage.begin(bookmark, txConfig);
+    const message = RequestMessage.begin(bookmark, txConfig, mode);
     this._connection.write(message, observer, true);
   }
 
@@ -70,8 +70,8 @@ export default class BoltProtocol extends BoltProtocolV2 {
     this._connection.write(message, observer, true);
   }
 
-  run(statement, parameters, bookmark, txConfig, observer) {
-    const runMessage = RequestMessage.runWithMetadata(statement, parameters, bookmark, txConfig);
+  run(statement, parameters, bookmark, txConfig, mode, observer) {
+    const runMessage = RequestMessage.runWithMetadata(statement, parameters, bookmark, txConfig, mode);
     const pullAllMessage = RequestMessage.pullAll();
 
     this._connection.write(runMessage, observer, false);

src/v1/internal/connection-holder.js

@@ -36,6 +36,14 @@ export default class ConnectionHolder {
     this._connectionPromise = Promise.resolve(null);
   }
 
+  /**
+   * Returns the assigned access mode.
+   * @returns {string} access mode
+   */
+  mode() {
+    return this._mode;
+  }
+
   /**
    * Make this holder initialize new connection if none exists already.
    * @return {undefined}

src/v1/internal/constants.js

@@ -0,0 +1,26 @@
+/**
+ * Copyright (c) 2002-2019 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+const ACCESS_MODE_READ = 'READ';
+const ACCESS_MODE_WRITE = 'WRITE';
+
+export {
+  ACCESS_MODE_READ,
+  ACCESS_MODE_WRITE
+}

src/v1/internal/request-message.js

@@ -17,6 +17,8 @@
  * limitations under the License.
  */
 
+import {ACCESS_MODE_READ} from './constants';
+
 // Signature bytes for each request message type
 const INIT = 0x01;            // 0000 0001 // INIT <user_agent> <authentication_token>
 const ACK_FAILURE = 0x0E;     // 0000 1110 // ACK_FAILURE - unused
@@ -31,6 +33,8 @@ const BEGIN = 0x11;           // 0001 0001 // BEGIN <metadata>
 const COMMIT = 0x12;          // 0001 0010 // COMMIT
 const ROLLBACK = 0x13;        // 0001 0011 // ROLLBACK
 
+const READ_MODE = "r";
+
 export default class RequestMessage {
 
   constructor(signature, fields, toString) {
@@ -90,10 +94,11 @@ export default class RequestMessage {
    * Create a new BEGIN message.
    * @param {Bookmark} bookmark the bookmark.
    * @param {TxConfig} txConfig the configuration.
+   * @param {string} mode the access mode.
    * @return {RequestMessage} new BEGIN message.
    */
-  static begin(bookmark, txConfig) {
-    const metadata = buildTxMetadata(bookmark, txConfig);
+  static begin(bookmark, txConfig, mode) {
+    const metadata = buildTxMetadata(bookmark, txConfig, mode);
     return new RequestMessage(BEGIN, [metadata], () => `BEGIN ${JSON.stringify(metadata)}`);
   }
 
@@ -119,10 +124,11 @@ export default class RequestMessage {
    * @param {object} parameters the statement parameters.
    * @param {Bookmark} bookmark the bookmark.
    * @param {TxConfig} txConfig the configuration.
+   * @param {string} mode the access mode.
    * @return {RequestMessage} new RUN message with additional metadata.
    */
-  static runWithMetadata(statement, parameters, bookmark, txConfig) {
-    const metadata = buildTxMetadata(bookmark, txConfig);
+  static runWithMetadata(statement, parameters, bookmark, txConfig, mode) {
+    const metadata = buildTxMetadata(bookmark, txConfig, mode);
     return new RequestMessage(RUN, [statement, parameters, metadata],
       () => `RUN ${statement} ${JSON.stringify(parameters)} ${JSON.stringify(metadata)}`);
   }
@@ -140,9 +146,10 @@ export default class RequestMessage {
  * Create an object that represent transaction metadata.
  * @param {Bookmark} bookmark the bookmark.
  * @param {TxConfig} txConfig the configuration.
+ * @param {string} mode the access mode.
  * @return {object} a metadata object.
  */
-function buildTxMetadata(bookmark, txConfig) {
+function buildTxMetadata(bookmark, txConfig, mode) {
   const metadata = {};
   if (!bookmark.isEmpty()) {
     metadata['bookmarks'] = bookmark.values();
@@ -153,6 +160,9 @@ function buildTxMetadata(bookmark, txConfig) {
   if (txConfig.metadata) {
     metadata['tx_metadata'] = txConfig.metadata;
   }
+  if (mode === ACCESS_MODE_READ) {
+    metadata['mode'] = READ_MODE;
+  }
   return metadata;
 }
 

src/v1/internal/routing-util.js

@@ -22,6 +22,7 @@ import Integer, {int} from '../integer';
 import {ServerVersion, VERSION_3_2_0} from './server-version';
 import Bookmark from './bookmark';
 import TxConfig from './tx-config';
+import {ACCESS_MODE_WRITE} from "./constants";
 
 const CALL_GET_SERVERS = 'CALL dbms.cluster.routing.getServers';
 const CALL_GET_ROUTING_TABLE = 'CALL dbms.cluster.routing.getRoutingTable($context)';
@@ -125,7 +126,7 @@ export default class RoutingUtil {
         params = {};
       }
 
-      connection.protocol().run(query, params, Bookmark.empty(), TxConfig.empty(), streamObserver);
+      connection.protocol().run(query, params, Bookmark.empty(), TxConfig.empty(), ACCESS_MODE_WRITE, streamObserver);
     });
   }
 }

src/v1/session.js

@@ -22,7 +22,8 @@ import Transaction from './transaction';
 import {newError} from './error';
 import {validateStatementAndParameters} from './internal/util';
 import ConnectionHolder from './internal/connection-holder';
-import Driver, {READ, WRITE} from './driver';
+import Driver from './driver';
+import {ACCESS_MODE_READ, ACCESS_MODE_WRITE} from './internal/constants';
 import TransactionExecutor from './internal/transaction-executor';
 import Bookmark from './internal/bookmark';
 import TxConfig from './internal/tx-config';
@@ -64,8 +65,8 @@ class Session {
    */
   constructor(mode, connectionProvider, bookmark, config) {
     this._mode = mode;
-    this._readConnectionHolder = new ConnectionHolder(READ, connectionProvider);
-    this._writeConnectionHolder = new ConnectionHolder(WRITE, connectionProvider);
+    this._readConnectionHolder = new ConnectionHolder(ACCESS_MODE_READ, connectionProvider);
+    this._writeConnectionHolder = new ConnectionHolder(ACCESS_MODE_WRITE, connectionProvider);
     this._open = true;
     this._hasTx = false;
     this._lastBookmark = bookmark;
@@ -86,7 +87,7 @@ class Session {
     const autoCommitTxConfig = transactionConfig ? new TxConfig(transactionConfig) : TxConfig.empty();
 
     return this._run(query, params, (connection, streamObserver) =>
-      connection.protocol().run(query, params, this._lastBookmark, autoCommitTxConfig, streamObserver)
+      connection.protocol().run(query, params, this._lastBookmark, autoCommitTxConfig, this._mode, streamObserver)
     );
   }
 
@@ -179,7 +180,7 @@ class Session {
    */
   readTransaction(transactionWork, transactionConfig) {
     const config = new TxConfig(transactionConfig);
-    return this._runTransaction(READ, config, transactionWork);
+    return this._runTransaction(ACCESS_MODE_READ, config, transactionWork);
   }
 
   /**
@@ -198,7 +199,7 @@ class Session {
    */
   writeTransaction(transactionWork, transactionConfig) {
     const config = new TxConfig(transactionConfig);
-    return this._runTransaction(WRITE, config, transactionWork);
+    return this._runTransaction(ACCESS_MODE_WRITE, config, transactionWork);
   }
 
   _runTransaction(accessMode, transactionConfig, transactionWork) {
@@ -238,9 +239,9 @@ class Session {
   }
 
   _connectionHolderWithMode(mode) {
-    if (mode === READ) {
+    if (mode === ACCESS_MODE_READ) {
       return this._readConnectionHolder;
-    } else if (mode === WRITE) {
+    } else if (mode === ACCESS_MODE_WRITE) {
       return this._writeConnectionHolder;
     } else {
       throw newError('Unknown access mode: ' + mode);

src/v1/transaction.js

@@ -46,7 +46,7 @@ class Transaction {
     const streamObserver = new _TransactionStreamObserver(this);
 
     this._connectionHolder.getConnection(streamObserver)
-      .then(conn => conn.protocol().beginTransaction(bookmark, txConfig, streamObserver))
+      .then(conn => conn.protocol().beginTransaction(bookmark, txConfig, this._connectionHolder.mode(), streamObserver))
       .catch(error => streamObserver.onError(error));
   }
 
@@ -158,7 +158,7 @@ let _states = {
       const txConfig = TxConfig.empty();
 
       connectionHolder.getConnection(observer)
-        .then(conn => conn.protocol().run(statement, parameters, bookmark, txConfig, observer))
+        .then(conn => conn.protocol().run(statement, parameters, bookmark, txConfig, connectionHolder.mode(), observer))
         .catch(error => observer.onError(error));
 
       return _newRunResult(observer, statement, parameters, () => observer.serverMetadata());

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

@@ -21,6 +21,7 @@ import BoltProtocolV1 from '../../src/v1/internal/bolt-protocol-v1';
 import RequestMessage from '../../src/v1/internal/request-message';
 import Bookmark from '../../src/v1/internal/bookmark';
 import TxConfig from '../../src/v1/internal/tx-config';
+import {WRITE} from "../../src/v1/driver";
 
 class MessageRecorder {
 
@@ -78,7 +79,7 @@ describe('BoltProtocolV1', () => {
     const parameters = {x: 'x', y: 'y'};
     const observer = {};
 
-    protocol.run(statement, parameters, Bookmark.empty(), TxConfig.empty(), observer);
+    protocol.run(statement, parameters, Bookmark.empty(), TxConfig.empty(), WRITE, observer);
 
     recorder.verifyMessageCount(2);
 
@@ -110,7 +111,7 @@ describe('BoltProtocolV1', () => {
     const bookmark = new Bookmark('neo4j:bookmark:v1:tx42');
     const observer = {};
 
-    protocol.beginTransaction(bookmark, TxConfig.empty(), observer);
+    protocol.beginTransaction(bookmark, TxConfig.empty(), WRITE, observer);
 
     recorder.verifyMessageCount(2);