Extract protocol negotiation into a dedicated class

It encapsulates all the logic related to the Bolt protocol
negotiation and versioned packstream components.

Author: lutovich

src/v1/internal/connector.js

@@ -20,14 +20,13 @@
 import WebSocketChannel from './ch-websocket';
 import NodeChannel from './ch-node';
 import {Chunker, Dechunker} from './chunking';
-import packStreamUtil from './packstream-util';
-import {alloc} from './buf';
 import {newError, PROTOCOL_ERROR} from './../error';
 import ChannelConfig from './ch-config';
 import urlUtil from './url-util';
 import StreamObserver from './stream-observer';
 import {ServerVersion, VERSION_3_2_0} from './server-version';
 import Logger from './logger';
+import ProtocolHandshaker from './protocol-handshaker';
 
 let Channel;
 if( NodeChannel.available ) {
@@ -41,21 +40,17 @@ else {
 }
 
 
-let
 // Signature bytes for each message type
-INIT = 0x01,            // 0000 0001 // INIT <user_agent>
-ACK_FAILURE = 0x0E,     // 0000 1110 // ACK_FAILURE - unused
-RESET = 0x0F,           // 0000 1111 // RESET
-RUN = 0x10,             // 0001 0000 // RUN <statement> <parameters>
-DISCARD_ALL = 0x2F,     // 0010 1111 // DISCARD *
-PULL_ALL = 0x3F,        // 0011 1111 // PULL *
-SUCCESS = 0x70,         // 0111 0000 // SUCCESS <metadata>
-RECORD = 0x71,          // 0111 0001 // RECORD <value>
-IGNORED = 0x7E,         // 0111 1110 // IGNORED <metadata>
-FAILURE = 0x7F,         // 0111 1111 // FAILURE <metadata>
-
-//sent before version negotiation
-MAGIC_PREAMBLE = 0x6060B017;
+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 *
+const SUCCESS = 0x70;         // 0111 0000 // SUCCESS <metadata>
+const RECORD = 0x71;          // 0111 0001 // RECORD <value>
+const IGNORED = 0x7E;         // 0111 1110 // IGNORED <metadata>
+const FAILURE = 0x7F;         // 0111 1111 // FAILURE <metadata>
 
 function NO_OP(){}
 
@@ -102,9 +97,11 @@ class Connection {
     this._chunker = new Chunker( channel );
     this._log = log;
 
+    const protocolHandshaker = new ProtocolHandshaker(this._ch, this._chunker, this._disableLosslessIntegers, this._log);
+
     // initially assume that database supports latest Bolt version, create latest packer and unpacker
-    this._packer = packStreamUtil.createLatestPacker(this._chunker);
-    this._unpacker = packStreamUtil.createLatestUnpacker(disableLosslessIntegers);
+    this._packer = protocolHandshaker.createLatestPacker();
+    this._unpacker = protocolHandshaker.createLatestUnpacker();
 
     this._currentFailure = null;
 
@@ -116,17 +113,7 @@ class Connection {
     // TODO: Using `onmessage` and `onerror` came from the WebSocket API,
     // it reads poorly and has several annoying drawbacks. Swap to having
     // Channel extend EventEmitter instead, then we can use `on('data',..)`
-    this._ch.onmessage = (buf) => {
-      const proposed = buf.readInt32();
-      if (proposed == 1 || proposed == 2) {
-        this._initializeProtocol(proposed, buf);
-      } else if (proposed == 1213486160) {//server responded 1213486160 == 0x48545450 == "HTTP"
-        this._handleFatalError(newError('Server responded HTTP. Make sure you are not trying to connect to the http endpoint ' +
-          '(HTTP defaults to port 7474 whereas BOLT defaults to port 7687)'));
-      } else {
-        this._handleFatalError(newError('Unknown Bolt protocol version: ' + proposed));
-      }
-    };
+    this._ch.onmessage = buffer => this._initializeProtocol(buffer, protocolHandshaker);
 
     // Listen to connection errors. Important note though;
     // In some cases we will get a channel that is already broken (for instance,
@@ -148,38 +135,31 @@ class Connection {
       this._log.debug(`${this} created towards ${hostPort}`);
     }
 
-    let handshake = alloc( 5 * 4 );
-    //magic preamble
-    handshake.writeInt32( MAGIC_PREAMBLE );
-    //proposed versions
-    handshake.writeInt32( 2 );
-    handshake.writeInt32( 1 );
-    handshake.writeInt32( 0 );
-    handshake.writeInt32( 0 );
-    handshake.reset();
-    this._ch.write( handshake );
+    protocolHandshaker.writeHandshakeRequest();
   }
 
   /**
    * Complete protocol initialization.
-   * @param {number} version the selected protocol version.
    * @param {BaseBuffer} buffer the handshake response buffer.
+   * @param {ProtocolHandshaker} protocolHandshaker the handshaker utility.
    * @private
    */
-  _initializeProtocol(version, buffer) {
-    if (this._log.isDebugEnabled()) {
-      this._log.debug(`${this} negotiated protocol version ${version}`);
-    }
+  _initializeProtocol(buffer, protocolHandshaker) {
+    try {
+      const {packer, unpacker} = protocolHandshaker.readHandshakeResponse(buffer);
 
-    // re-create packer and unpacker because version might be lower than we initially assumed
-    this._packer = packStreamUtil.createPackerForProtocolVersion(version, this._chunker);
-    this._unpacker = packStreamUtil.createUnpackerForProtocolVersion(version, this._disableLosslessIntegers);
+      // re-assign packer and unpacker because version might be lower than we initially assumed
+      this._packer = packer;
+      this._unpacker = unpacker;
 
-    // Ok, protocol running. Simply forward all messages to the dechunker
-    this._ch.onmessage = buf => this._dechunker.write(buf);
+      // Ok, protocol running. Simply forward all messages to the dechunker
+      this._ch.onmessage = buf => this._dechunker.write(buf);
 
-    if (buffer.hasRemaining()) {
-      this._dechunker.write(buffer.readSlice(buffer.remaining()));
+      if (buffer.hasRemaining()) {
+        this._dechunker.write(buffer.readSlice(buffer.remaining()));
+      }
+    } catch (e) {
+      this._handleFatalError(e);
     }
   }
 

src/v1/internal/packstream-util.js

@@ -0,0 +1,107 @@
+/**
+ * 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 {alloc} from './buf';
+import {newError} from '../error';
+import * as v1 from './packstream-v1';
+import * as v2 from './packstream-v2';
+
+const HTTP_MAGIC_PREAMBLE = 1213486160; // == 0x48545450 == "HTTP"
+const BOLT_MAGIC_PREAMBLE = 0x6060B017;
+
+const PACKER_CONSTRUCTORS_BY_VERSION = [null, v1.Packer, v2.Packer];
+const UNPACKER_CONSTRUCTORS_BY_VERSION = [null, v1.Unpacker, v2.Unpacker];
+
+export default class ProtocolHandshaker {
+
+  constructor(channel, chunker, disableLosslessIntegers, log) {
+    this._channel = channel;
+    this._chunker = chunker;
+    this._disableLosslessIntegers = disableLosslessIntegers;
+    this._log = log;
+  }
+
+  createLatestPacker() {
+    return this._createPackerForProtocolVersion(PACKER_CONSTRUCTORS_BY_VERSION.length - 1);
+  }
+
+  createLatestUnpacker() {
+    return this._createUnpackerForProtocolVersion(UNPACKER_CONSTRUCTORS_BY_VERSION.length - 1);
+  }
+
+  writeHandshakeRequest() {
+    this._channel.write(newHandshakeBuffer());
+  }
+
+  readHandshakeResponse(buffer) {
+    const proposedVersion = buffer.readInt32();
+
+    if (proposedVersion === 1 || proposedVersion === 2) {
+      return this._createPackerAndUnpackerForProtocolVersion(proposedVersion);
+    } else if (proposedVersion === HTTP_MAGIC_PREAMBLE) {
+      throw newError('Server responded HTTP. Make sure you are not trying to connect to the http endpoint ' +
+        '(HTTP defaults to port 7474 whereas BOLT defaults to port 7687)');
+    } else {
+      throw newError('Unknown Bolt protocol version: ' + proposedVersion);
+    }
+  }
+
+  _createPackerAndUnpackerForProtocolVersion(version) {
+    if (this._log.isDebugEnabled()) {
+      this._log.debug(`${this} negotiated protocol version ${version}`);
+    }
+    const packer = this._createPackerForProtocolVersion(version);
+    const unpacker = this._createUnpackerForProtocolVersion(version);
+    return {packer, unpacker};
+  }
+
+  _createPackerForProtocolVersion(version) {
+    const packerConstructor = PACKER_CONSTRUCTORS_BY_VERSION[version];
+    if (!packerConstructor) {
+      throw new Error(`Packer can't be created for protocol version ${version}`);
+    }
+    return new packerConstructor(this._chunker);
+  }
+
+  _createUnpackerForProtocolVersion(version) {
+    const unpackerConstructor = UNPACKER_CONSTRUCTORS_BY_VERSION[version];
+    if (!unpackerConstructor) {
+      throw new Error(`Unpacker can't be created for protocol version ${version}`);
+    }
+    return new unpackerConstructor(this._disableLosslessIntegers);
+  }
+}
+
+function newHandshakeBuffer() {
+  const handshakeBuffer = alloc(5 * 4);
+
+  //magic preamble
+  handshakeBuffer.writeInt32(BOLT_MAGIC_PREAMBLE);
+
+  //proposed versions
+  handshakeBuffer.writeInt32(2);
+  handshakeBuffer.writeInt32(1);
+  handshakeBuffer.writeInt32(0);
+  handshakeBuffer.writeInt32(0);
+
+  // reset the reader position
+  handshakeBuffer.reset();
+
+  return handshakeBuffer;
+}

src/v1/internal/protocol-handshaker.js

@@ -17,34 +17,37 @@
  * limitations under the License.
  */
 
-import packStreamUtil from '../../src/v1/internal/packstream-util';
 import * as v1 from '../../src/v1/internal/packstream-v1';
 import * as v2 from '../../src/v1/internal/packstream-v2';
+import ProtocolHandshaker from '../../src/v1/internal/protocol-handshaker';
+import Logger from '../../src/v1/internal/logger';
 
-describe('packstream-util', () => {
+describe('ProtocolHandshaker', () => {
+
+  const protocolHandshaker = new ProtocolHandshaker(null, false, Logger.noOp());
 
   it('should create packer of the specified version', () => {
-    const packer1 = packStreamUtil.createPackerForProtocolVersion(1, null);
+    const packer1 = protocolHandshaker._createPackerForProtocolVersion(1);
     expect(packer1 instanceof v1.Packer).toBeTruthy();
 
-    const packer2 = packStreamUtil.createPackerForProtocolVersion(2, null);
+    const packer2 = protocolHandshaker._createPackerForProtocolVersion(2);
     expect(packer2 instanceof v2.Packer).toBeTruthy();
   });
 
   it('should create unpacker of the specified version', () => {
-    const unpacker1 = packStreamUtil.createUnpackerForProtocolVersion(1, null);
+    const unpacker1 = protocolHandshaker._createUnpackerForProtocolVersion(1);
     expect(unpacker1 instanceof v1.Unpacker).toBeTruthy();
 
-    const unpacker2 = packStreamUtil.createUnpackerForProtocolVersion(2, null);
+    const unpacker2 = protocolHandshaker._createUnpackerForProtocolVersion(2);
     expect(unpacker2 instanceof v2.Unpacker).toBeTruthy();
   });
 
   it('should fail to create packer for unknown version', () => {
-    expect(() => packStreamUtil.createPackerForProtocolVersion(42, null)).toThrow();
+    expect(() => protocolHandshaker._createPackerForProtocolVersion(42)).toThrow();
   });
 
   it('should fail to create unpacker for unknown version', () => {
-    expect(() => packStreamUtil.createUnpackerForProtocolVersion(42, null)).toThrow();
+    expect(() => protocolHandshaker._createUnpackerForProtocolVersion(42)).toThrow();
   });
 
 });