Refactoring to better support new protocol versions

Connection is now less coupled with protocol messages. It is able
to write an arbitrary `RequestMessage` instead of exposing functions
to write messages of a specific type, like `Connection#pullAll()`.
Also introduced a `BoltProtocol` abstraction which exposes only logical
operations, like "initialize the connection" or "run a query". It
should make it simpler to introduce new protocol version that uses
different messages to execute same logical operations.

Author: lutovich

src/v1/driver.js

@@ -132,9 +132,9 @@ class Driver {
    * @access private
    */
   _createConnection(hostPort, release) {
-    let conn = connect(hostPort, this._config, this._connectionErrorCode(), this._log);
-    let streamObserver = new _ConnectionStreamObserver(this, conn);
-    conn.initialize(this._userAgent, this._token, streamObserver);
+    const conn = connect(hostPort, this._config, this._connectionErrorCode(), this._log);
+    const streamObserver = new _ConnectionStreamObserver(this, conn);
+    conn.protocol().initialize(this._userAgent, this._token, streamObserver);
     conn._release = () => release(hostPort, conn);
 
     this._openConnections[conn.id] = conn;

src/v1/internal/bolt-protocol.js

@@ -0,0 +1,84 @@
+/**
+ * Copyright (c) 2002-2018 "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.
+ */
+import RequestMessage from './request-message';
+
+export default class BoltProtocol {
+
+  /**
+   * @constructor
+   * @param {Connection} connection the connection.
+   * @param {Packer} packer the packer.
+   * @param {Unpacker} unpacker the unpacker.
+   */
+  constructor(connection, packer, unpacker) {
+    this._connection = connection;
+    this._packer = packer;
+    this._unpacker = unpacker;
+  }
+
+  /**
+   * Get the packer.
+   * @return {Packer} the protocol's packer.
+   */
+  packer() {
+    return this._packer;
+  }
+
+  /**
+   * Get the unpacker.
+   * @return {Unpacker} the protocol's unpacker.
+   */
+  unpacker() {
+    return this._unpacker;
+  }
+
+  /**
+   * Perform initialization and authentication of the underlying connection.
+   * @param {string} clientName the client name.
+   * @param {object} authToken the authentication token.
+   * @param {StreamObserver} observer the response observer.
+   */
+  initialize(clientName, authToken, observer) {
+    const message = RequestMessage.init(clientName, authToken);
+    this._connection.write(message, observer, true);
+  }
+
+  /**
+   * Send a Cypher statement through the underlying connection.
+   * @param {string} statement the cypher statement.
+   * @param {object} parameters the statement parameters.
+   * @param {StreamObserver} observer the response observer.
+   */
+  run(statement, parameters, observer) {
+    const runMessage = RequestMessage.run(statement, parameters);
+    const pullAllMessage = RequestMessage.pullAll();
+
+    this._connection.write(runMessage, observer, false);
+    this._connection.write(pullAllMessage, observer, true);
+  }
+
+  /**
+   * Send a RESET through the underlying connection.
+   * @param {StreamObserver} observer the response observer.
+   */
+  reset(observer) {
+    const message = RequestMessage.reset();
+    this._connection.write(message, observer, true);
+  }
+}

src/v1/internal/buf.js

@@ -578,11 +578,10 @@ try {
 } catch(e) {}
 
 /**
- * Allocate a new buffer using whatever mechanism is most sensible for the
- * current platform
+ * Allocate a new buffer using whatever mechanism is most sensible for the current platform.
  * @access private
  * @param {Integer} size
- * @return new buffer
+ * @return {BaseBuffer} new buffer
  */
 function alloc (size) {
   return new _DefaultBuffer(size);

src/v1/internal/connector.js

@@ -27,6 +27,7 @@ import StreamObserver from './stream-observer';
 import {ServerVersion, VERSION_3_2_0} from './server-version';
 import Logger from './logger';
 import ProtocolHandshaker from './protocol-handshaker';
+import RequestMessage from './request-message';
 
 let Channel;
 if( NodeChannel.available ) {
@@ -40,13 +41,7 @@ else {
 }
 
 
-// Signature bytes for each message type
-const INIT = 0x01;            // 0000 0001 // INIT <user_agent>
-const ACK_FAILURE = 0x0E;     // 0000 1110 // ACK_FAILURE - unused
-const RESET = 0x0F;           // 0000 1111 // RESET
-const RUN = 0x10;             // 0001 0000 // RUN <statement> <parameters>
-const DISCARD_ALL = 0x2F;     // 0010 1111 // DISCARD *
-const PULL_ALL = 0x3F;        // 0011 1111 // PULL *
+// Signature bytes for each response message type
 const SUCCESS = 0x70;         // 0111 0000 // SUCCESS <metadata>
 const RECORD = 0x71;          // 0111 0001 // RECORD <value>
 const IGNORED = 0x7E;         // 0111 1110 // IGNORED <metadata>
@@ -97,11 +92,9 @@ class Connection {
     this._chunker = new Chunker( channel );
     this._log = log;
 
-    const protocolHandshaker = new ProtocolHandshaker(this._ch, this._chunker, this._disableLosslessIntegers, this._log);
-
+    const protocolHandshaker = new ProtocolHandshaker(this, channel, this._chunker, this._disableLosslessIntegers, this._log);
     // initially assume that database supports latest Bolt version, create latest packer and unpacker
-    this._packer = protocolHandshaker.createLatestPacker();
-    this._unpacker = protocolHandshaker.createLatestUnpacker();
+    this._protocol = protocolHandshaker.createLatestProtocol();
 
     this._currentFailure = null;
 
@@ -128,7 +121,7 @@ class Connection {
     }
 
     this._dechunker.onmessage = (buf) => {
-      this._handleMessage(this._unpacker.unpack(buf));
+      this._handleMessage(this._protocol.unpacker().unpack(buf));
     };
 
     if (this._log.isDebugEnabled()) {
@@ -138,6 +131,45 @@ class Connection {
     protocolHandshaker.writeHandshakeRequest();
   }
 
+  /**
+   * Get the Bolt protocol for the connection.
+   * @return {BoltProtocol} the protocol.
+   */
+  protocol() {
+    return this._protocol;
+  }
+
+  /**
+   * Write a message to the network channel.
+   * @param {RequestMessage} message the message to write.
+   * @param {StreamObserver} observer the response observer.
+   * @param {boolean} flush <code>true</code> if flush should happen after the message is written to the buffer.
+   */
+  write(message, observer, flush) {
+    if (message.isInitializationMessage) {
+      observer = this._state.wrap(observer);
+    }
+
+    const queued = this._queueObserver(observer);
+
+    if (queued) {
+      if (this._log.isDebugEnabled()) {
+        this._log.debug(`${this} C: ${message}`);
+      }
+
+      this._protocol.packer().packStruct(
+        message.signature,
+        message.fields.map(field => this._packable(field)),
+        err => this._handleFatalError(err));
+
+      this._chunker.messageBoundary();
+
+      if (flush) {
+        this._chunker.flush();
+      }
+    }
+  }
+
   /**
    * Complete protocol initialization.
    * @param {BaseBuffer} buffer the handshake response buffer.
@@ -146,11 +178,8 @@ class Connection {
    */
   _initializeProtocol(buffer, protocolHandshaker) {
     try {
-      const {packer, unpacker} = protocolHandshaker.readHandshakeResponse(buffer);
-
-      // re-assign packer and unpacker because version might be lower than we initially assumed
-      this._packer = packer;
-      this._unpacker = unpacker;
+      // re-assign the protocol because version might be lower than we initially assumed
+      this._protocol = protocolHandshaker.readHandshakeResponse(buffer);
 
       // Ok, protocol running. Simply forward all messages to the dechunker
       this._ch.onmessage = buf => this._dechunker.write(buf);
@@ -247,66 +276,13 @@ class Connection {
     }
   }
 
-  /** Queue an INIT-message to be sent to the database */
-  initialize( clientName, token, observer ) {
-    if (this._log.isDebugEnabled()) {
-      this._log.debug(`${this} C: INIT ${clientName} {...}`);
-    }
-    const initObserver = this._state.wrap(observer);
-    const queued = this._queueObserver(initObserver);
-    if (queued) {
-      this._packer.packStruct(INIT, [this._packable(clientName), this._packable(token)],
-        (err) => this._handleFatalError(err));
-      this._chunker.messageBoundary();
-      this.flush();
-    }
-  }
-
-  /** Queue a RUN-message to be sent to the database */
-  run( statement, params, observer ) {
-    if (this._log.isDebugEnabled()) {
-      this._log.debug(`${this} C: RUN ${statement} ${JSON.stringify(params)}`);
-    }
-    const queued = this._queueObserver(observer);
-    if (queued) {
-      this._packer.packStruct(RUN, [this._packable(statement), this._packable(params)],
-        (err) => this._handleFatalError(err));
-      this._chunker.messageBoundary();
-    }
-  }
-
-  /** Queue a PULL_ALL-message to be sent to the database */
-  pullAll( observer ) {
-    if (this._log.isDebugEnabled()) {
-      this._log.debug(`${this} C: PULL_ALL`);
-    }
-    const queued = this._queueObserver(observer);
-    if (queued) {
-      this._packer.packStruct(PULL_ALL, [], (err) => this._handleFatalError(err));
-      this._chunker.messageBoundary();
-    }
-  }
-
-  /** Queue a DISCARD_ALL-message to be sent to the database */
-  discardAll( observer ) {
-    if (this._log.isDebugEnabled()) {
-      this._log.debug(`${this} C: DISCARD_ALL`);
-    }
-    const queued = this._queueObserver(observer);
-    if (queued) {
-      this._packer.packStruct(DISCARD_ALL, [], (err) => this._handleFatalError(err));
-      this._chunker.messageBoundary();
-    }
-  }
-
   /**
-   * Send a RESET-message to the database. Mutes failure handling.
-   * Message is immediately flushed to the network. Separate {@link Connection#flush()} call is not required.
+   * Send a RESET-message to the database. Message is immediately flushed to the network.
    * @return {Promise<void>} promise resolved when SUCCESS-message response arrives, or failed when other response messages arrives.
    */
   resetAndFlush() {
     return new Promise((resolve, reject) => {
-      this._reset({
+      this._protocol.reset({
         onNext: record => {
           const neo4jError = this._handleProtocolError('Received RECORD as a response for RESET: ' + JSON.stringify(record));
           reject(neo4jError);
@@ -328,7 +304,7 @@ class Connection {
   }
 
   _resetOnFailure() {
-    this._reset({
+    this._protocol.reset({
       onNext: record => {
         this._handleProtocolError('Received RECORD as a response for RESET: ' + JSON.stringify(record));
       },
@@ -342,19 +318,6 @@ class Connection {
     });
   }
 
-  _reset(observer) {
-    if (this._log.isDebugEnabled()) {
-      this._log.debug(`${this} C: RESET`);
-    }
-
-    const queued = this._queueObserver(observer);
-    if (queued) {
-      this._packer.packStruct(RESET, [], err => this._handleFatalError(err));
-      this._chunker.messageBoundary();
-      this.flush();
-    }
-  }
-
   _queueObserver(observer) {
     if( this._isBroken ) {
       if( observer && observer.onError ) {
@@ -376,7 +339,7 @@ class Connection {
 
   /**
    * Get promise resolved when connection initialization succeed or rejected when it fails.
-   * Connection is initialized using {@link initialize} function.
+   * Connection is initialized using {@link BoltProtocol#initialize()} function.
    * @return {Promise<Connection>} the result of connection initialization.
    */
   initializationCompleted() {
@@ -391,13 +354,6 @@ class Connection {
     this._currentObserver = this._pendingObservers.shift();
   }
 
-  /**
-   * Flush all queued outgoing messages.
-   */
-  flush() {
-    this._chunker.flush();
-  }
-
   /** Check if this connection is in working condition */
   isOpen() {
     return !this._isBroken && this._ch._open;
@@ -419,7 +375,7 @@ class Connection {
   }
 
   _packable(value) {
-      return this._packer.packable(value, (err) => this._handleFatalError(err));
+    return this._protocol.packer().packable(value, (err) => this._handleFatalError(err));
   }
 
   /**
@@ -431,7 +387,7 @@ class Connection {
       this.server.version = serverVersion;
       const version = ServerVersion.fromString(serverVersion);
       if (version.compareTo(VERSION_3_2_0) < 0) {
-        this._packer.disableByteArrays();
+        this._protocol.packer().disableByteArrays();
       }
     }
   }