moving the handshake to before creating connection

Author: bigmontz

src/internal/connection-channel.js

@@ -21,10 +21,14 @@ import { Channel } from './node'
 import { Chunker, Dechunker } from './chunking'
 import { newError, PROTOCOL_ERROR } from '../error'
 import ChannelConfig from './channel-config'
-import ProtocolHandshaker from './protocol-handshaker'
+import ProtocolHandshaker, {
+  newHandshakeBuffer,
+  parseNegotiatedResponse
+} from './protocol-handshaker'
 import Connection from './connection'
 import BoltProtocol from './bolt-protocol-v1'
 import { ResultStreamObserver } from './stream-observers'
+import buffer from 'vinyl-buffer'
 
 // Signature bytes for each response message type
 const SUCCESS = 0x70 // 0111 0000 // SUCCESS <metadata>
@@ -42,6 +46,118 @@ const NO_OP_OBSERVER = {
 
 let idGenerator = 0
 
+/**
+ * Shake hands using the channel and return the protocol version
+ *
+ * @param {Channel} channel the channel use to shake hands
+ * @returns {Promise<{}>} Promise of protocol version and consumeRemainingBuffer
+ */
+function shakeHands (channel) {
+  return new Promise((resolve, reject) => {
+    const handshakeErrorHandler = error => {
+      reject(error)
+    }
+
+    channel.onerror = handshakeErrorHandler.bind(this)
+    if (channel._error) {
+      // channel is already broken
+      handshakeErrorHandler(channel._error)
+    }
+
+    channel.onmessage = buffer => {
+      try {
+        // read the response buffer and initialize the protocol
+        const protocolVersion = parseNegotiatedResponse(buffer)
+
+        resolve({
+          protocolVersion,
+          consumeRemainingBuffer: consumer => {
+            if (buffer.hasRemaining()) {
+              consumer(buffer.readSlice(buffer.remaining()))
+            }
+          }
+        })
+      } catch (e) {
+        this._handleFatalError(e)
+        reject(e)
+      }
+    }
+
+    channel.write(newHandshakeBuffer())
+  })
+}
+
+/**
+ * Crete new connection to the provided address. Returned connection is not connected.
+ * @param {ServerAddress} address - the Bolt endpoint to connect to.
+ * @param {Object} config - the driver configuration.
+ * @param {ConnectionErrorHandler} errorHandler - the error handler for connection errors.
+ * @param {Logger} log - configured logger.
+ * @return {Connection} - new connection.
+ */
+export function createChannelConnection (
+  address,
+  config,
+  errorHandler,
+  log,
+  serversideRouting = null
+) {
+  const channelConfig = new ChannelConfig(
+    address,
+    config,
+    errorHandler.errorCode()
+  )
+
+  const channel = new Channel(channelConfig)
+
+  return shakeHands(channel)
+    .then(({ protocolVersion, consumeRemainingBuffer }) => {
+      const connection = new ChannelConnection(
+        channel,
+        errorHandler,
+        address,
+        log,
+        config.disableLosslessIntegers,
+        serversideRouting
+      )
+
+      // Temporary, this object will not be used in the feature
+      const protocolHandshaker = new ProtocolHandshaker(
+        connection,
+        channel,
+        connection._chunker,
+        connection._disableLosslessIntegers,
+        connection._log,
+        connection._serversideRouting
+      )
+
+      connection._protocol = protocolHandshaker._createProtocolWithVersion(
+        protocolVersion
+      )
+
+      // reset the error handler to just handle errors and forget about the handshake promise
+      channel.onerror = connection._handleFatalError.bind(connection)
+
+      // Ok, protocol running. Simply forward all messages to the dechunker
+      channel.onmessage = buf => connection._dechunker.write(buf)
+
+      // setup dechunker to dechunk messages and forward them to the message handler
+      connection._dechunker.onmessage = buf => {
+        connection._handleMessage(connection._protocol.unpacker().unpack(buf))
+      }
+
+      // forward all pending bytes to the dechunker
+      consumeRemainingBuffer(buffer => connection._dechunker.write(buffer))
+
+      return connection
+    })
+    .catch(reason =>
+      channel.close().then(() => {
+        throw reason
+      })
+    )
+}
+
 export default class ChannelConnection extends Connection {
   /**
    * @constructor
@@ -148,6 +264,10 @@ export default class ChannelConnection extends Connection {
    * @return {Promise<Connection>} promise resolved with the current connection if handshake is successful. Rejected promise otherwise.
    */
   _negotiateProtocol () {
+    if (this._protocol) {
+      return Promise.resolve(this)
+    }
+
     const protocolHandshaker = new ProtocolHandshaker(
       this,
       this._ch,
@@ -262,12 +382,10 @@ export default class ChannelConnection extends Connection {
         this._log.debug(`${this} C: ${message}`)
       }
 
-      this._protocol
-        .packer()
-        .packStruct(
-          message.signature,
-          message.fields.map(field => this._packable(field))
-        )
+      this._protocol.packer().packStruct(
+        message.signature,
+        message.fields.map(field => this._packable(field))
+      )
 
       this._chunker.messageBoundary()
 

src/internal/connection-provider-direct.js

@@ -19,7 +19,7 @@
 
 import PooledConnectionProvider from './connection-provider-pooled'
 import DelegateConnection from './connection-delegate'
-import ChannelConnection from './connection-channel'
+import { createChannelConnection } from './connection-channel'
 import { BOLT_PROTOCOL_V4_0, BOLT_PROTOCOL_V3 } from './constants'
 
 export default class DirectConnectionProvider extends PooledConnectionProvider {
@@ -40,25 +40,24 @@ export default class DirectConnectionProvider extends PooledConnectionProvider {
   }
 
   async _hasProtocolVersion (versionPredicate) {
-    const connection = ChannelConnection.create(
+    const connection = await createChannelConnection(
       this._address,
       this._config,
       this._createConnectionErrorHandler(),
       this._log
     )
 
-    try {
-      await connection._negotiateProtocol()
+    const protocolVersion = connection.protocol()
+      ? connection.protocol().version
+      : null
 
-      const protocol = connection.protocol()
-      if (protocol) {
-        return versionPredicate(protocol.version)
-      }
+    await connection.close()
 
-      return false
-    } finally {
-      await connection.close()
+    if (protocolVersion) {
+      return versionPredicate(protocolVersion)
     }
+
+    return false
   }
 
   async supportsMultiDb () {

src/internal/connection-provider-pooled.js

@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 
-import ChannelConnection from './connection-channel'
+import { createChannelConnection } from './connection-channel'
 import Pool from './pool'
 import PoolConfig from './pool-config'
 import ConnectionErrorHandler from './connection-error-handler'
@@ -39,7 +39,7 @@ export default class PooledConnectionProvider extends ConnectionProvider {
     this._createChannelConnection =
       createChannelConnectionHook ||
       (address => {
-        return ChannelConnection.create(
+        return createChannelConnection(
           address,
           this._config,
           this._createConnectionErrorHandler(),
@@ -72,15 +72,17 @@ export default class PooledConnectionProvider extends ConnectionProvider {
    * @access private
    */
   _createConnection (address, release) {
-    const connection = this._createChannelConnection(address)
-    connection._release = () => release(address, connection)
-    this._openConnections[connection.id] = connection
-
-    return connection.connect(this._userAgent, this._authToken).catch(error => {
-      // let's destroy this connection
-      this._destroyConnection(connection)
-      // propagate the error because connection failed to connect / initialize
-      throw error
+    return this._createChannelConnection(address).then(connection => {
+      connection._release = () => release(address, connection)
+      this._openConnections[connection.id] = connection
+      return connection
+        .connect(this._userAgent, this._authToken)
+        .catch(error => {
+          // let's destroy this connection
+          this._destroyConnection(connection)
+          // propagate the error because connection failed to connect / initialize
+          throw error
+        })
     })
   }
 

src/internal/connection-provider-routing.js

@@ -29,7 +29,7 @@ import ConnectionErrorHandler from './connection-error-handler'
 import DelegateConnection from './connection-delegate'
 import LeastConnectedLoadBalancingStrategy from './least-connected-load-balancing-strategy'
 import Bookmark from './bookmark'
-import ChannelConnection from './connection-channel'
+import { createChannelConnection } from './connection-channel'
 import { int } from '../integer'
 import { BOLT_PROTOCOL_V3, BOLT_PROTOCOL_V4_0 } from './constants'
 
@@ -53,7 +53,7 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
     routingTablePurgeDelay
   }) {
     super({ id, config, log, userAgent, authToken }, address => {
-      return ChannelConnection.create(
+      return createChannelConnection(
         address,
         this._config,
         this._createConnectionErrorHandler(),
@@ -164,27 +164,30 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
 
     let lastError
     for (let i = 0; i < addresses.length; i++) {
-      const connection = ChannelConnection.create(
-        addresses[i],
-        this._config,
-        this._createConnectionErrorHandler(),
-        this._log
-      )
-
       try {
-        await connection._negotiateProtocol()
+        const connection = await createChannelConnection(
+          this._address,
+          this._config,
+          this._createConnectionErrorHandler(),
+          this._log
+        )
 
-        const protocol = connection.protocol()
-        if (protocol) {
-          return versionPredicate(protocol.version)
+        const protocolVersion = connection.protocol()
+          ? connection.protocol().version
+          : null
+
+        await connection.close()
+
+        if (protocolVersion) {
+          return versionPredicate(protocolVersion)
         }
 
         return false
       } catch (error) {
         lastError = error
-      } finally {
-        await connection.close()
       }
+
+      return false
     }
 
     if (lastError) {

src/internal/protocol-handshaker.js

@@ -71,7 +71,7 @@ function createHandshakeMessage (versions) {
   return handshakeBuffer
 }
 
-function parseNegotiatedResponse (buffer) {
+export function parseNegotiatedResponse (buffer) {
   const h = [
     buffer.readUInt8(),
     buffer.readUInt8(),
@@ -118,10 +118,7 @@ export default class ProtocolHandshaker {
    * Write a Bolt handshake into the underlying network channel.
    */
   writeHandshakeRequest () {
-    this._chunker.write(newHandshakeBuffer())
-    this._chunker.messageBoundary()
-    this._chunker.flush()
-    // this._channel.write(newHandshakeBuffer())
+    this._channel.write(newHandshakeBuffer())
   }
 
   /**
@@ -201,7 +198,7 @@ export default class ProtocolHandshaker {
  * @return {BaseBuffer}
  * @private
  */
-function newHandshakeBuffer () {
+export function newHandshakeBuffer () {
   return createHandshakeMessageMemoized([
     [BOLT_PROTOCOL_V4_3, BOLT_PROTOCOL_V4_2],
     BOLT_PROTOCOL_V4_1,