broken: extract response handler from the connection

Author: bigmontz

src/internal/bolt/index.js

@@ -1,7 +1,9 @@
 import handshake from './handshake'
 import create from './create'
+import ResponseHandler from './response-handler'
 
 export default {
   handshake,
-  create
+  create,
+  ResponseHandler
 }

src/internal/bolt/response-handler.js

@@ -0,0 +1,214 @@
+/**
+ * Copyright (c) 2002-2020 "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 { newError, PROTOCOL_ERROR } from '../../error'
+
+// 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>
+const FAILURE = 0x7f // 0111 1111 // FAILURE <metadata>
+
+function NO_OP () {}
+
+const NO_OP_OBSERVER = {
+  onNext: NO_OP,
+  onCompleted: NO_OP,
+  onError: NO_OP
+}
+
+export default class ResponseHandler {
+  constructor ({ protocol, channel, dechunker, log, errorHandler } = {}) {
+    this._pendingObservers = []
+    this._currentObserver = undefined
+    this._isBroken = false
+    this._log = log
+    this._protocol = protocol // should gone from here
+    this._errorHandler = errorHandler
+
+    // reset the error handler to just handle errors and forget about the handshake promise
+    channel.onerror = this._handleFatalError.bind(this)
+
+    // Ok, protocol running. Simply forward all messages to the dechunker
+    channel.onmessage = buf => dechunker.write(buf)
+
+    // setup dechunker to dechunk messages and forward them to the message handler
+    dechunker.onmessage = buf => {
+      this._handleMessage(protocol.unpacker().unpack(buf))
+    }
+  }
+
+  queueObserver (observer) {
+    if (this._isBroken) {
+      if (observer && observer.onError) {
+        observer.onError(this._error)
+      }
+      return false
+    }
+    observer = observer || NO_OP_OBSERVER
+    observer.onCompleted = observer.onCompleted || NO_OP
+    observer.onError = observer.onError || NO_OP
+    observer.onNext = observer.onNext || NO_OP
+    if (this._currentObserver === undefined) {
+      this._currentObserver = observer
+    } else {
+      this._pendingObservers.push(observer)
+    }
+    return true
+  }
+
+  handleProtocolError (message) {
+    this._currentFailure = null
+    this._updateCurrentObserver()
+    const error = newError(message, PROTOCOL_ERROR)
+    this._handleFatalError(error)
+    return error
+  }
+
+  get isBroken () {
+    return this._isBroken
+  }
+
+  /*
+   * Pop next pending observer form the list of observers and make it current observer.
+   * @protected
+   */
+  _updateCurrentObserver () {
+    this._currentObserver = this._pendingObservers.shift()
+  }
+
+  _handleMessage (msg) {
+    if (this._isBroken) {
+      // ignore all incoming messages when this connection is broken. all previously pending observers failed
+      // with the fatal error. all future observers will fail with same fatal error.
+      return
+    }
+
+    const payload = msg.fields[0]
+
+    switch (msg.signature) {
+      case RECORD:
+        if (this._log.isDebugEnabled()) {
+          this._log.debug(`${this} S: RECORD ${JSON.stringify(msg)}`)
+        }
+        this._currentObserver.onNext(payload)
+        break
+      case SUCCESS:
+        if (this._log.isDebugEnabled()) {
+          this._log.debug(`${this} S: SUCCESS ${JSON.stringify(msg)}`)
+        }
+        try {
+          const metadata = this._protocol.transformMetadata(payload)
+          this._currentObserver.onCompleted(metadata)
+        } finally {
+          this._updateCurrentObserver()
+        }
+        break
+      case FAILURE:
+        if (this._log.isDebugEnabled()) {
+          this._log.debug(`${this} S: FAILURE ${JSON.stringify(msg)}`)
+        }
+        try {
+          const error = newError(payload.message, payload.code)
+          this._currentFailure = this.handleAndTransformError(
+            error,
+            this._address
+          )
+          this._currentObserver.onError(this._currentFailure)
+        } finally {
+          this._updateCurrentObserver()
+          // Things are now broken. Pending observers will get FAILURE messages routed until we are done handling this failure.
+          this._resetOnFailure()
+        }
+        break
+      case IGNORED:
+        if (this._log.isDebugEnabled()) {
+          this._log.debug(`${this} S: IGNORED ${JSON.stringify(msg)}`)
+        }
+        try {
+          if (this._currentFailure && this._currentObserver.onError) {
+            this._currentObserver.onError(this._currentFailure)
+          } else if (this._currentObserver.onError) {
+            this._currentObserver.onError(
+              newError('Ignored either because of an error or RESET')
+            )
+          }
+        } finally {
+          this._updateCurrentObserver()
+        }
+        break
+      default:
+        this._handleFatalError(
+          newError('Unknown Bolt protocol message: ' + msg)
+        )
+    }
+  }
+
+  /**
+   * "Fatal" means the connection is dead. Only call this if something
+   * happens that cannot be recovered from. This will lead to all subscribers
+   * failing, and the connection getting ejected from the session pool.
+   *
+   * @param error an error object, forwarded to all current and future subscribers
+   */
+  _handleFatalError (error) {
+    this._isBroken = true
+    this._error = this.handleAndTransformError(error, this._address)
+
+    if (this._log.isErrorEnabled()) {
+      this._log.error(
+        `${this} experienced a fatal error ${JSON.stringify(this._error)}`
+      )
+    }
+
+    if (this._currentObserver && this._currentObserver.onError) {
+      this._currentObserver.onError(this._error)
+    }
+    while (this._pendingObservers.length > 0) {
+      const observer = this._pendingObservers.shift()
+      if (observer && observer.onError) {
+        observer.onError(this._error)
+      }
+    }
+  }
+
+  _resetOnFailure () {
+    this._protocol.reset({
+      onError: () => {
+        this._currentFailure = null
+      },
+      onComplete: () => {
+        this._currentFailure = null
+      }
+    })
+  }
+
+  /**
+   *
+   * @param error
+   * @param address
+   * @returns {Neo4jError|*}
+   */
+  handleAndTransformError (error, address) {
+    if (this._errorHandler) {
+      return this._errorHandler.handleAndTransformError(error, address)
+    }
+
+    return error
+  }
+}