Fix connection receive timeout for single requests (#1166)

Resets and non-pipeline requests were not triggering the connection receive timeout.
The problem happened because of the timeout mechanics only started when pending observers are set.
However, the current observer should be also be considered for this use case.

Author: bigmontz

packages/bolt-connection/src/bolt/response-handler.js

@@ -76,7 +76,7 @@ export default class ResponseHandler {
     this._transformMetadata = transformMetadata || NO_OP_IDENTITY
     this._observer = Object.assign(
       {
-        onPendingObserversChange: NO_OP,
+        onObserversCountChange: NO_OP,
         onError: NO_OP,
         onFailure: NO_OP,
         onErrorApplyTransformation: NO_OP_IDENTITY
@@ -156,7 +156,11 @@ export default class ResponseHandler {
    */
   _updateCurrentObserver () {
     this._currentObserver = this._pendingObservers.shift()
-    this._observer.onPendingObserversChange(this._pendingObservers.length)
+    this._observer.onObserversCountChange(this._observersCount)
+  }
+
+  get _observersCount () {
+    return this._currentObserver == null ? this._pendingObservers.length : this._pendingObservers.length + 1
   }
 
   _queueObserver (observer) {
@@ -169,7 +173,7 @@ export default class ResponseHandler {
     } else {
       this._pendingObservers.push(observer)
     }
-    this._observer.onPendingObserversChange(this._pendingObservers.length)
+    this._observer.onObserversCountChange(this._observersCount)
     return true
   }
 

packages/bolt-connection/src/connection/connection-channel.js

@@ -67,7 +67,7 @@ export function createChannelConnection (
           server: conn.server,
           log: conn.logger,
           observer: {
-            onPendingObserversChange: conn._handleOngoingRequestsNumberChange.bind(conn),
+            onObserversCountChange: conn._handleOngoingRequestsNumberChange.bind(conn),
             onError: conn._handleFatalError.bind(conn),
             onFailure: conn._resetOnFailure.bind(conn),
             onProtocolError: conn._handleProtocolError.bind(conn),

packages/bolt-connection/test/bolt/response-handler.test.js

@@ -22,6 +22,7 @@ const {
   logger: { Logger }
 } = internal
 
+const SUCCESS = 0x70 // 0111 0000 // SUCCESS <metadata>
 const FAILURE = 0x7f // 0111 1111 // FAILURE <metadata>
 
 describe('response-handler', () => {
@@ -69,4 +70,36 @@ describe('response-handler', () => {
       expect(receivedError.code).toBe(expectedError.code)
     })
   })
+
+  it('should keep track of observers and notify onObserversCountChange()', () => {
+    const observer = {
+      onObserversCountChange: jest.fn()
+    }
+    const responseHandler = new ResponseHandler({ observer, log: Logger.noOp() })
+
+    responseHandler._queueObserver({})
+    expect(observer.onObserversCountChange).toHaveBeenLastCalledWith(1)
+
+    responseHandler._queueObserver({})
+    expect(observer.onObserversCountChange).toHaveBeenLastCalledWith(2)
+
+    responseHandler._queueObserver({})
+    expect(observer.onObserversCountChange).toHaveBeenLastCalledWith(3)
+
+    const success = {
+      signature: SUCCESS,
+      fields: [{}]
+    }
+
+    responseHandler.handleResponse(success)
+    expect(observer.onObserversCountChange).toHaveBeenLastCalledWith(2)
+
+    responseHandler.handleResponse(success)
+    expect(observer.onObserversCountChange).toHaveBeenLastCalledWith(1)
+
+    responseHandler.handleResponse(success)
+    expect(observer.onObserversCountChange).toHaveBeenLastCalledWith(0)
+
+    expect(observer.onObserversCountChange).toHaveBeenCalledTimes(6)
+  })
 })

packages/neo4j-driver-deno/lib/bolt-connection/bolt/response-handler.js

@@ -76,7 +76,7 @@ export default class ResponseHandler {
     this._transformMetadata = transformMetadata || NO_OP_IDENTITY
     this._observer = Object.assign(
       {
-        onPendingObserversChange: NO_OP,
+        onObserversCountChange: NO_OP,
         onError: NO_OP,
         onFailure: NO_OP,
         onErrorApplyTransformation: NO_OP_IDENTITY
@@ -156,7 +156,11 @@ export default class ResponseHandler {
    */
   _updateCurrentObserver () {
     this._currentObserver = this._pendingObservers.shift()
-    this._observer.onPendingObserversChange(this._pendingObservers.length)
+    this._observer.onObserversCountChange(this._observersCount)
+  }
+
+  get _observersCount () {
+    return this._currentObserver == null ? this._pendingObservers.length : this._pendingObservers.length + 1
   }
 
   _queueObserver (observer) {
@@ -169,7 +173,7 @@ export default class ResponseHandler {
     } else {
       this._pendingObservers.push(observer)
     }
-    this._observer.onPendingObserversChange(this._pendingObservers.length)
+    this._observer.onObserversCountChange(this._observersCount)
     return true
   }
 

packages/neo4j-driver-deno/lib/bolt-connection/connection/connection-channel.js

@@ -67,7 +67,7 @@ export function createChannelConnection (
           server: conn.server,
           log: conn.logger,
           observer: {
-            onPendingObserversChange: conn._handleOngoingRequestsNumberChange.bind(conn),
+            onObserversCountChange: conn._handleOngoingRequestsNumberChange.bind(conn),
             onError: conn._handleFatalError.bind(conn),
             onFailure: conn._resetOnFailure.bind(conn),
             onProtocolError: conn._handleProtocolError.bind(conn),