Bolt V2 negotiation

Allow driver to negotiate version 2 of the Bolt protocol and select
`Packer`/`Unpacker` based on the agreed protocol version. Struct
unpacking logic moved to the V1 unpacker. Driver initially tries to
newest available `Packer`/`Unpacker` and downgrades them after the
handshake response, if needed.

Author: lutovich

src/v1/internal/ch-dummy.js

@@ -58,11 +58,15 @@ class DummyChannel {
   }
 
   close(cb) {
-    this.written = [];
+    this.clear();
     if (cb) {
       return cb();
     }
   }
+
+  clear() {
+    this.written = [];
+  }
 }
 
 const channel = DummyChannel;

src/v1/internal/connector.js

@@ -20,9 +20,8 @@
 import WebSocketChannel from './ch-websocket';
 import NodeChannel from './ch-node';
 import {Chunker, Dechunker} from './chunking';
-import {Packer, Unpacker} from './packstream';
+import packStreamUtil from './packstream-util';
 import {alloc} from './buf';
-import {Node, Path, PathSegment, Relationship, UnboundRelationship} from '../graph-types';
 import {newError} from './../error';
 import ChannelConfig from './ch-config';
 import urlUtil from './url-util';
@@ -54,11 +53,6 @@ RECORD = 0x71,          // 0111 0001 // RECORD <value>
 IGNORED = 0x7E,         // 0111 1110 // IGNORED <metadata>
 FAILURE = 0x7F,         // 0111 1111 // FAILURE <metadata>
 
-// Signature bytes for higher-level graph objects
-NODE = 0x4E,
-RELATIONSHIP = 0x52,
-UNBOUND_RELATIONSHIP = 0x72,
-PATH = 0x50,
 //sent before version negotiation
 MAGIC_PREAMBLE = 0x6060B017,
 DEBUG = false;
@@ -85,66 +79,6 @@ let NO_OP_OBSERVER = {
   onError : NO_OP
 };
 
-/** Maps from packstream structures to Neo4j domain objects */
-let _mappers = {
-  node : ( unpacker, buf ) => {
-    return new Node(
-      unpacker.unpack(buf), // Identity
-      unpacker.unpack(buf), // Labels
-      unpacker.unpack(buf)  // Properties
-    );
-  },
-  rel : ( unpacker, buf ) => {
-    return new Relationship(
-      unpacker.unpack(buf),  // Identity
-      unpacker.unpack(buf),  // Start Node Identity
-      unpacker.unpack(buf),  // End Node Identity
-      unpacker.unpack(buf),  // Type
-      unpacker.unpack(buf) // Properties
-    );
-  },
-  unboundRel : ( unpacker, buf ) => {
-    return new UnboundRelationship(
-      unpacker.unpack(buf),  // Identity
-      unpacker.unpack(buf),  // Type
-      unpacker.unpack(buf) // Properties
-    );
-  },
-  path : ( unpacker, buf ) => {
-    let nodes = unpacker.unpack(buf),
-        rels = unpacker.unpack(buf),
-        sequence = unpacker.unpack(buf);
-    let prevNode = nodes[0],
-        segments = [];
-
-    for (let i = 0; i < sequence.length; i += 2) {
-        let relIndex = sequence[i],
-            nextNode = nodes[sequence[i + 1]],
-            rel;
-        if (relIndex > 0) {
-            rel = rels[relIndex - 1];
-            if( rel instanceof UnboundRelationship ) {
-              // To avoid duplication, relationships in a path do not contain
-              // information about their start and end nodes, that's instead
-              // inferred from the path sequence. This is us inferring (and,
-              // for performance reasons remembering) the start/end of a rel.
-              rels[relIndex - 1] = rel = rel.bind(prevNode.identity, nextNode.identity);
-            }
-        } else {
-            rel = rels[-relIndex - 1];
-            if( rel instanceof UnboundRelationship ) {
-              // See above
-              rels[-relIndex - 1] = rel = rel.bind(nextNode.identity, prevNode.identity);
-            }
-        }
-        // Done hydrating one path segment.
-        segments.push( new PathSegment( prevNode, rel, nextNode ) );
-        prevNode = nextNode;
-    }
-    return new Path(nodes[0], nodes[nodes.length - 1],  segments );
-  }
-};
-
 /**
  * A connection manages sending and recieving messages over a channel. A
  * connector is very closely tied to the Bolt protocol, it implements the
@@ -175,13 +109,16 @@ class Connection {
     this.url = url;
     this.server = {address: url};
     this.creationTimestamp = Date.now();
+    this._disableLosslessIntegers = disableLosslessIntegers;
     this._pendingObservers = [];
     this._currentObserver = undefined;
     this._ch = channel;
     this._dechunker = new Dechunker();
     this._chunker = new Chunker( channel );
-    this._packer = new Packer(this._chunker);
-    this._unpacker = new Unpacker(disableLosslessIntegers);
+
+    // 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._isHandlingFailure = false;
     this._currentFailure = null;
@@ -191,34 +128,18 @@ class Connection {
     // Set to true on fatal errors, to get this out of session pool.
     this._isBroken = false;
 
-    // For deserialization, explain to the unpacker how to unpack nodes, rels, paths;
-    this._unpacker.structMappers[NODE] = _mappers.node;
-    this._unpacker.structMappers[RELATIONSHIP] = _mappers.rel;
-    this._unpacker.structMappers[UNBOUND_RELATIONSHIP] = _mappers.unboundRel;
-    this._unpacker.structMappers[PATH] = _mappers.path;
-
-    let self = this;
     // 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) => {
-      let proposed = buf.readInt32();
-      if( proposed == 1 ) {
-        // Ok, protocol running. Simply forward all messages past
-        // this to the dechunker
-        self._ch.onmessage = (buf) => {
-          self._dechunker.write(buf);
-        };
-
-        if( buf.hasRemaining() ) {
-          self._dechunker.write(buf.readSlice( buf.remaining() ));
-        }
+      const proposed = buf.readInt32();
+      if (proposed == 1 || proposed == 2) {
+        this._initializeProtocol(proposed, buf);
       } else if (proposed == 1213486160) {//server responded 1213486160 == 0x48545450 == "HTTP"
-        self._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 {
-        self._handleFatalError(newError("Unknown Bolt protocol version: " + proposed));
+        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));
       }
     };
 
@@ -235,21 +156,40 @@ class Connection {
     }
 
     this._dechunker.onmessage = (buf) => {
-      self._handleMessage( self._unpacker.unpack( buf ) );
+      this._handleMessage(this._unpacker.unpack(buf));
     };
 
     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.writeInt32( 0 );
     handshake.reset();
     this._ch.write( handshake );
   }
 
+  /**
+   * Complete protocol initialization.
+   * @param {number} version the selected protocol version.
+   * @param {BaseBuffer} buffer the handshake response buffer.
+   * @private
+   */
+  _initializeProtocol(version, 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);
+
+    // 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()));
+    }
+  }
+
   /**
    * "Fatal" means the connection is dead. Only call this if something
    * happens that cannot be recovered from. This will lead to all subscribers

src/v1/internal/packstream-util.js

@@ -0,0 +1,55 @@
+/**
+ * Copyright (c) 2002-2018 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.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 * as v1 from './packstream-v1';
+import * as v2 from './packstream-v2';
+
+const PACKER_CONSTRUCTORS_BY_VERSION = [null, v1.Packer, v2.Packer];
+const UNPACKER_CONSTRUCTORS_BY_VERSION = [null, v1.Unpacker, v2.Unpacker];
+
+function createLatestPacker(chunker) {
+  return createPackerForProtocolVersion(PACKER_CONSTRUCTORS_BY_VERSION.length - 1, chunker);
+}
+
+function createLatestUnpacker(disableLosslessIntegers) {
+  return createUnpackerForProtocolVersion(UNPACKER_CONSTRUCTORS_BY_VERSION.length - 1, disableLosslessIntegers);
+}
+
+function createPackerForProtocolVersion(version, chunker) {
+  const packerConstructor = PACKER_CONSTRUCTORS_BY_VERSION[version];
+  if (!packerConstructor) {
+    throw new Error(`Packer can't be created for protocol version ${version}`);
+  }
+  return new packerConstructor(chunker);
+}
+
+function createUnpackerForProtocolVersion(version, disableLosslessIntegers) {
+  const unpackerConstructor = UNPACKER_CONSTRUCTORS_BY_VERSION[version];
+  if (!unpackerConstructor) {
+    throw new Error(`Unpacker can't be created for protocol version ${version}`);
+  }
+  return new unpackerConstructor(disableLosslessIntegers);
+}
+
+export default {
+  createLatestPacker: createLatestPacker,
+  createLatestUnpacker: createLatestUnpacker,
+  createPackerForProtocolVersion: createPackerForProtocolVersion,
+  createUnpackerForProtocolVersion: createUnpackerForProtocolVersion
+};

src/v1/internal/packstream.js renamed to src/v1/internal/packstream-v1.js

@@ -20,6 +20,7 @@ import utf8 from './utf8';
 import Integer, {int, isInt} from '../integer';
 import {newError} from './../error';
 import {Chunker} from './chunking';
+import {Node, Path, PathSegment, Relationship, UnboundRelationship} from '../graph-types';
 
 const TINY_STRING = 0x80;
 const TINY_LIST = 0x90;
@@ -48,6 +49,11 @@ const MAP_32 = 0xDA;
 const STRUCT_8 = 0xDC;
 const STRUCT_16 = 0xDD;
 
+const NODE = 0x4E;
+const RELATIONSHIP = 0x52;
+const UNBOUND_RELATIONSHIP = 0x72;
+const PATH = 0x50;
+
 /**
   * A Structure have a signature and fields.
   * @access private
@@ -322,10 +328,6 @@ class Unpacker {
    * @param {boolean} disableLosslessIntegers if this unpacker should convert all received integers to native JS numbers.
    */
   constructor(disableLosslessIntegers = false) {
-    // Higher level layers can specify how to map structs to higher-level objects.
-    // If we receive a struct that has a signature that does not have a mapper,
-    // we simply return a Structure object.
-    this.structMappers = {};
     this._disableLosslessIntegers = disableLosslessIntegers;
   }
 
@@ -504,17 +506,88 @@ class Unpacker {
   }
 
   _unpackStructWithSize(size, buffer) {
-    let signature = buffer.readUInt8();
-    let mapper = this.structMappers[signature];
-    if (mapper) {
-      return mapper(this, buffer);
+    const signature = buffer.readUInt8();
+    if (signature == NODE) {
+      return this._unpackNode(buffer);
+    } else if (signature == RELATIONSHIP) {
+      return this._unpackRelationship(buffer);
+    } else if (signature == UNBOUND_RELATIONSHIP) {
+      return this._unpackUnboundRelationship(buffer);
+    } else if (signature == PATH) {
+      return this._unpackPath(buffer);
     } else {
-      let value = new Structure(signature, []);
-      for (let i = 0; i < size; i++) {
-        value.fields.push(this.unpack(buffer));
+      return this._unpackUnknownStruct(signature, size, buffer);
+    }
+  }
+
+  _unpackNode(buffer) {
+    return new Node(
+      this.unpack(buffer), // Identity
+      this.unpack(buffer), // Labels
+      this.unpack(buffer)  // Properties
+    );
+  }
+
+  _unpackRelationship(buffer) {
+    return new Relationship(
+      this.unpack(buffer), // Identity
+      this.unpack(buffer), // Start Node Identity
+      this.unpack(buffer), // End Node Identity
+      this.unpack(buffer), // Type
+      this.unpack(buffer)  // Properties
+    );
+  }
+
+  _unpackUnboundRelationship(buffer) {
+    return new UnboundRelationship(
+      this.unpack(buffer), // Identity
+      this.unpack(buffer), // Type
+      this.unpack(buffer)  // Properties
+    );
+  }
+
+  _unpackPath(buffer) {
+    const nodes = this.unpack(buffer);
+    const rels = this.unpack(buffer);
+    const sequence = this.unpack(buffer);
+
+    const segments = [];
+    let prevNode = nodes[0];
+
+    for (let i = 0; i < sequence.length; i += 2) {
+      const nextNode = nodes[sequence[i + 1]];
+      let relIndex = sequence[i];
+      let rel;
+
+      if (relIndex > 0) {
+        rel = rels[relIndex - 1];
+        if (rel instanceof UnboundRelationship) {
+          // To avoid duplication, relationships in a path do not contain
+          // information about their start and end nodes, that's instead
+          // inferred from the path sequence. This is us inferring (and,
+          // for performance reasons remembering) the start/end of a rel.
+          rels[relIndex - 1] = rel = rel.bind(prevNode.identity, nextNode.identity);
+        }
+      } else {
+        rel = rels[-relIndex - 1];
+        if (rel instanceof UnboundRelationship) {
+          // See above
+          rels[-relIndex - 1] = rel = rel.bind(nextNode.identity, prevNode.identity);
+        }
       }
-      return value;
+      // Done hydrating one path segment.
+      segments.push(new PathSegment(prevNode, rel, nextNode));
+      prevNode = nextNode;
+    }
+    return new Path(nodes[0], nodes[nodes.length - 1], segments);
+  }
+
+  _unpackUnknownStruct(signature, size, buffer) {
+    const result = new Structure(signature, []);
+    for (let i = 0; i < size; i++) {
+      result.fields.push(this.unpack(buffer));
     }
+    return result;
   }
 
 }