response-handler: remove part of the connection dependency

Author: bigmontz

src/internal/bolt/response-handler.js

@@ -17,6 +17,7 @@
  * limitations under the License.
  */
 import { newError, PROTOCOL_ERROR } from '../../error'
+import BoltProtocol from '../bolt-protocol-v1'
 
 // Signature bytes for each response message type
 const SUCCESS = 0x70 // 0111 0000 // SUCCESS <metadata>
@@ -32,16 +33,48 @@ const NO_OP_OBSERVER = {
   onError: NO_OP
 }
 
+/**
+ * Treat the protocol responses and notify the observers
+ */
 export default class ResponseHandler {
-  constructor ({ protocol, channel, dechunker, log, connection } = {}) {
+  /**
+   * Called when something went wrong with the connectio
+   * @callback ResponseHandler~Observer~OnError
+   * @param {any} error The error
+   */
+  /**
+   * @typedef {Object} ResponseHandler~Observer
+   * @property {ResponseHandler~Observer~OnError} onError Invoke when a connection error occurs
+   * @property {ResponseHandler~Observer~OnError} onFailure Invoke when a protocol failure occurs
+   */
+  /**
+   * Constructor
+   * @param {Object} param The params
+   * @param {BoltProtocol} protocol Bolt protocol used to parse message (to be removed)
+   * @param {Channel} channel The channel used to exchange messages
+   * @param {Logger} log The logger
+   * @param {ResponseHandler~Observer} observer Object which will be notified about errors
+   */
+  constructor ({
+    protocol,
+    channel,
+    dechunker,
+    log,
+    observer,
+    connection
+  } = {}) {
     this._pendingObservers = []
     this._isBroken = false
     this._log = log
     this._protocol = protocol // should gone from here
     this._connection = connection // should gone from here
+    this._observer = Object.assign(
+      { onError: NO_OP, onFailure: NO_OP },
+      observer
+    )
 
     // reset the error handler to just handle errors and forget about the handshake promise
-    channel.onerror = connection._handleFatalError.bind(connection)
+    channel.onerror = this._observer.onError.bind(this._observer)
 
     // Ok, protocol running. Simply forward all messages to the dechunker
     channel.onmessage = buf => dechunker.write(buf)
@@ -53,11 +86,11 @@ export default class ResponseHandler {
   }
 
   _handleMessage (msg) {
-    if (this._connection._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
-    }
+    // if (this._connection._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]
 
@@ -93,7 +126,7 @@ export default class ResponseHandler {
         } finally {
           this._updateCurrentObserver()
           // Things are now broken. Pending observers will get FAILURE messages routed until we are done handling this failure.
-          this._connection._resetOnFailure()
+          this._observer.onFailure(this._currentFailure)
         }
         break
       case IGNORED:
@@ -113,7 +146,7 @@ export default class ResponseHandler {
         }
         break
       default:
-        this._connection._handleFatalError(
+        this._observer.onError(
           newError('Unknown Bolt protocol message: ' + msg)
         )
     }
@@ -128,12 +161,6 @@ export default class ResponseHandler {
   }
 
   _queueObserver (observer) {
-    if (this._connection._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

src/internal/connection-channel.js

@@ -84,6 +84,10 @@ export function createChannelConnection (
         channel,
         dechunker,
         log,
+        observer: {
+          onError: connection._handleFatalError.bind(connection),
+          onFailure: connection._resetOnFailure.bind(connection)
+        },
         connection
       })
 
@@ -334,6 +338,12 @@ export default class ChannelConnection extends Connection {
   }
 
   _queueObserver (observer) {
+    if (this._isBroken) {
+      if (observer && observer.onError) {
+        observer.onError(this._error)
+      }
+      return false
+    }
     return this._responseHandler._queueObserver(observer)
   }