Fix connection getting timeout while idle

packages/bolt-connection/src/connection-provider/connection-provider-pooled.js

@@ -226,11 +226,11 @@ export default class PooledConnectionProvider extends ConnectionProvider {
   }
 
   static _installIdleObserverOnConnection (conn, observer) {
-    conn._queueObserver(observer)
+    conn._setIdle(observer)
   }
 
   static _removeIdleObserverOnConnection (conn) {
-    conn._updateCurrentObserver()
+    conn._unsetIdle()
   }
 
   _handleSecurityError (error, address, connection) {

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

@@ -125,6 +125,7 @@ export default class ChannelConnection extends Connection {
   ) {
     super(errorHandler)
     this._authToken = null
+    this._idle = false
     this._reseting = false
     this._resetObservers = []
     this._id = idGenerator++
@@ -393,7 +394,26 @@ export default class ChannelConnection extends Connection {
   }
 
   /**
-   * This method still here because it's used by the {@link PooledConnectionProvider}
+   * This method is used by the {@link PooledConnectionProvider}
+   *
+   * @param {any} observer
+   */
+  _setIdle (observer) {
+    this._idle = true
+    this._ch.stopReceiveTimeout()
+    this._protocol.queueObserverIfProtocolIsNotBroken(observer)
+  }
+
+  /**
+   * This method is used by the {@link PooledConnectionProvider}
+   */
+  _unsetIdle () {
+    this._idle = false
+    this._updateCurrentObserver()
+  }
+
+  /**
+   * This method still here because of the connection-channel.tests.js
    *
    * @param {any} observer
    */
@@ -402,7 +422,7 @@ export default class ChannelConnection extends Connection {
   }
 
   hasOngoingObservableRequests () {
-    return this._protocol.hasOngoingObservableRequests()
+    return !this._idle && this._protocol.hasOngoingObservableRequests()
   }
 
   /**
@@ -500,6 +520,9 @@ export default class ChannelConnection extends Connection {
    * @param {number} requestsNumber Ongoing requests number
    */
   _handleOngoingRequestsNumberChange (requestsNumber) {
+    if (this._idle) {
+      return
+    }
     if (requestsNumber === 0) {
       this._ch.stopReceiveTimeout()
     } else {

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

@@ -559,7 +559,7 @@ describe('ChannelConnection', () => {
   })
 
   describe('.__handleOngoingRequestsNumberChange()', () => {
-    it('should call channel.stopReceiveTimeout when requets number equals to 0', () => {
+    it('should call channel.stopReceiveTimeout when requests number equals to 0', () => {
       const channel = {
         stopReceiveTimeout: jest.fn().mockName('stopReceiveTimeout'),
         startReceiveTimeout: jest.fn().mockName('startReceiveTimeout')
@@ -571,7 +571,7 @@ describe('ChannelConnection', () => {
       expect(channel.stopReceiveTimeout).toHaveBeenCalledTimes(1)
     })
 
-    it('should not call channel.startReceiveTimeout when requets number equals to 0', () => {
+    it('should not call channel.startReceiveTimeout when requests number equals to 0', () => {
       const channel = {
         stopReceiveTimeout: jest.fn().mockName('stopReceiveTimeout'),
         startReceiveTimeout: jest.fn().mockName('startReceiveTimeout')
@@ -585,7 +585,7 @@ describe('ChannelConnection', () => {
 
     it.each([
       [1], [2], [3], [5], [8], [13], [3000]
-    ])('should call channel.startReceiveTimeout when requets number equals to %d', (requests) => {
+    ])('should call channel.startReceiveTimeout when requests number equals to %d', (requests) => {
       const channel = {
         stopReceiveTimeout: jest.fn().mockName('stopReceiveTimeout'),
         startReceiveTimeout: jest.fn().mockName('startReceiveTimeout')
@@ -599,7 +599,7 @@ describe('ChannelConnection', () => {
 
     it.each([
       [1], [2], [3], [5], [8], [13], [3000]
-    ])('should not call channel.stopReceiveTimeout when requets number equals to %d', (requests) => {
+    ])('should not call channel.stopReceiveTimeout when requests number equals to %d', (requests) => {
       const channel = {
         stopReceiveTimeout: jest.fn().mockName('stopReceiveTimeout'),
         startReceiveTimeout: jest.fn().mockName('startReceiveTimeout')
@@ -610,6 +610,68 @@ describe('ChannelConnection', () => {
 
       expect(channel.stopReceiveTimeout).toHaveBeenCalledTimes(0)
     })
+
+    it.each([
+      [0], [1], [2], [3], [5], [8], [13], [3000]
+    ])('should not call channel.stopReceiveTimeout or startReceiveTimeout when requests number equals to %d and connection is idle', (requests) => {
+      const channel = {
+        stopReceiveTimeout: jest.fn().mockName('stopReceiveTimeout'),
+        startReceiveTimeout: jest.fn().mockName('startReceiveTimeout')
+      }
+      const protocol = {
+        queueObserverIfProtocolIsNotBroken: jest.fn(() => {})
+      }
+      const connection = spyOnConnectionChannel({ channel, protocolSupplier: () => protocol })
+      connection._setIdle({})
+      channel.stopReceiveTimeout.mockClear()
+
+      connection._handleOngoingRequestsNumberChange(requests)
+
+      expect(channel.stopReceiveTimeout).toHaveBeenCalledTimes(0)
+      expect(channel.startReceiveTimeout).toHaveBeenCalledTimes(0)
+    })
+
+    it.each([
+      [1], [2], [3], [5], [8], [13], [3000]
+    ])('should  call channel.startReceiveTimeout when requests number equals to %d and connection is not idle anymore', (requests) => {
+      const channel = {
+        stopReceiveTimeout: jest.fn().mockName('stopReceiveTimeout'),
+        startReceiveTimeout: jest.fn().mockName('startReceiveTimeout')
+      }
+      const protocol = {
+        queueObserverIfProtocolIsNotBroken: jest.fn(() => {}),
+        updateCurrentObserver: jest.fn(() => {})
+      }
+      const connection = spyOnConnectionChannel({ channel, protocolSupplier: () => protocol })
+      connection._setIdle({})
+      connection._unsetIdle()
+      channel.stopReceiveTimeout.mockClear()
+
+      connection._handleOngoingRequestsNumberChange(requests)
+
+      expect(channel.stopReceiveTimeout).toHaveBeenCalledTimes(0)
+      expect(channel.startReceiveTimeout).toHaveBeenCalledTimes(1)
+    })
+
+    it('should  call channel.stopReceiveTimeout when requests number equals to 0 and connection is not idle anymore', () => {
+      const channel = {
+        stopReceiveTimeout: jest.fn().mockName('stopReceiveTimeout'),
+        startReceiveTimeout: jest.fn().mockName('startReceiveTimeout')
+      }
+      const protocol = {
+        queueObserverIfProtocolIsNotBroken: jest.fn(() => {}),
+        updateCurrentObserver: jest.fn(() => {})
+      }
+      const connection = spyOnConnectionChannel({ channel, protocolSupplier: () => protocol })
+      connection._setIdle({})
+      connection._unsetIdle()
+      channel.stopReceiveTimeout.mockClear()
+
+      connection._handleOngoingRequestsNumberChange(0)
+
+      expect(channel.stopReceiveTimeout).toHaveBeenCalledTimes(1)
+      expect(channel.startReceiveTimeout).toHaveBeenCalledTimes(0)
+    })
   })
 
   describe('.resetAndFlush()', () => {
@@ -1181,6 +1243,44 @@ describe('ChannelConnection', () => {
   })
 
   describe('.hasOngoingObservableRequests()', () => {
+    it('should return false if connection is idle', () => {
+      const protocol = {
+        hasOngoingObservableRequests: jest.fn(() => true),
+        queueObserverIfProtocolIsNotBroken: jest.fn(() => {})
+      }
+      const channel = {
+        stopReceiveTimeout: jest.fn().mockName('stopReceiveTimeout')
+      }
+
+      const connection = spyOnConnectionChannel({ protocolSupplier: () => protocol, channel })
+      connection._setIdle({})
+
+      const result = connection.hasOngoingObservableRequests()
+
+      expect(result).toBe(false)
+      expect(protocol.hasOngoingObservableRequests).not.toBeCalledWith()
+    })
+
+    it('should redirect request to the protocol when connection is not idle anymore', () => {
+      const protocol = {
+        hasOngoingObservableRequests: jest.fn(() => true),
+        queueObserverIfProtocolIsNotBroken: jest.fn(() => {}),
+        updateCurrentObserver: jest.fn(() => {})
+      }
+      const channel = {
+        stopReceiveTimeout: jest.fn().mockName('stopReceiveTimeout')
+      }
+
+      const connection = spyOnConnectionChannel({ protocolSupplier: () => protocol, channel })
+      connection._setIdle({})
+      connection._unsetIdle()
+
+      const result = connection.hasOngoingObservableRequests()
+
+      expect(result).toBe(true)
+      expect(protocol.hasOngoingObservableRequests).toBeCalledWith()
+    })
+
     it('should call redirect request to the protocol', () => {
       const protocol = {
         hasOngoingObservableRequests: jest.fn(() => true)
@@ -1195,6 +1295,41 @@ describe('ChannelConnection', () => {
     })
   })
 
+  describe('._setIdle()', () => {
+    it('should stop receive timeout and enqueue observer', () => {
+      const protocol = {
+        queueObserverIfProtocolIsNotBroken: jest.fn(() => {})
+      }
+      const channel = {
+        stopReceiveTimeout: jest.fn().mockName('stopReceiveTimeout')
+      }
+      const observer = {
+        onComplete: () => {}
+      }
+
+      const connection = spyOnConnectionChannel({ protocolSupplier: () => protocol, channel })
+
+      connection._setIdle(observer)
+
+      expect(channel.stopReceiveTimeout).toBeCalledTimes(1)
+      expect(protocol.queueObserverIfProtocolIsNotBroken).toBeCalledWith(observer)
+    })
+  })
+
+  describe('._unsetIdle()', () => {
+    it('should update current observer', () => {
+      const protocol = {
+        updateCurrentObserver: jest.fn(() => {})
+      }
+
+      const connection = spyOnConnectionChannel({ protocolSupplier: () => protocol })
+
+      connection._unsetIdle()
+
+      expect(protocol.updateCurrentObserver).toBeCalledTimes(1)
+    })
+  })
+
   function spyOnConnectionChannel ({
     channel,
     errorHandler,

packages/neo4j-driver-deno/lib/bolt-connection/connection-provider/connection-provider-pooled.js

@@ -226,11 +226,11 @@ export default class PooledConnectionProvider extends ConnectionProvider {
   }
 
   static _installIdleObserverOnConnection (conn, observer) {
-    conn._queueObserver(observer)
+    conn._setIdle(observer)
   }
 
   static _removeIdleObserverOnConnection (conn) {
-    conn._updateCurrentObserver()
+    conn._unsetIdle()
   }
 
   _handleSecurityError (error, address, connection) {

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

@@ -125,6 +125,7 @@ export default class ChannelConnection extends Connection {
   ) {
     super(errorHandler)
     this._authToken = null
+    this._idle = false
     this._reseting = false
     this._resetObservers = []
     this._id = idGenerator++
@@ -393,7 +394,26 @@ export default class ChannelConnection extends Connection {
   }
 
   /**
-   * This method still here because it's used by the {@link PooledConnectionProvider}
+   * This method is used by the {@link PooledConnectionProvider}
+   *
+   * @param {any} observer
+   */
+  _setIdle (observer) {
+    this._idle = true
+    this._ch.stopReceiveTimeout()
+    this._protocol.queueObserverIfProtocolIsNotBroken(observer)
+  }
+
+  /**
+   * This method is used by the {@link PooledConnectionProvider}
+   */
+  _unsetIdle () {
+    this._idle = false
+    this._updateCurrentObserver()
+  }
+
+  /**
+   * This method still here because of the connection-channel.tests.js
    *
    * @param {any} observer
    */
@@ -402,7 +422,7 @@ export default class ChannelConnection extends Connection {
   }
 
   hasOngoingObservableRequests () {
-    return this._protocol.hasOngoingObservableRequests()
+    return !this._idle && this._protocol.hasOngoingObservableRequests()
   }
 
   /**
@@ -500,6 +520,9 @@ export default class ChannelConnection extends Connection {
    * @param {number} requestsNumber Ongoing requests number
    */
   _handleOngoingRequestsNumberChange (requestsNumber) {
+    if (this._idle) {
+      return
+    }
     if (requestsNumber === 0) {
       this._ch.stopReceiveTimeout()
     } else {

packages/neo4j-driver/test/internal/connection-provider-pooled.test.js

@@ -71,4 +71,29 @@ describe('#unit PooledConnectionProvider', () => {
       clock.uninstall()
     }
   })
+
+  it('_installIdleObserverOnConnection should set connection as idle', () => {
+    const connection = new FakeConnection()
+    const observer = { onCompleted: () => {} }
+
+    PooledConnectionProvider._installIdleObserverOnConnection(connection, observer)
+
+    expect(connection._idle).toBe(true)
+    expect(connection._idleObserver).toBe(observer)
+  })
+
+  it('_removeIdleObserverOnConnection should unset connection as idle', () => {
+    const connection = new FakeConnection()
+    const observer = { onCompleted: () => {} }
+
+    PooledConnectionProvider._installIdleObserverOnConnection(connection, observer)
+
+    expect(connection._idle).toBe(true)
+    expect(connection._idleObserver).toBe(observer)
+
+    PooledConnectionProvider._removeIdleObserverOnConnection(connection)
+
+    expect(connection._idle).toBe(false)
+    expect(connection._idleObserver).toBe(null)
+  })
 })

packages/neo4j-driver/test/internal/fake-connection.js

@@ -39,7 +39,8 @@ export default class FakeConnection extends Connection {
     this._databaseId = null
     this._requestRoutingInformationMock = null
     this._creationTimestamp = Date.now()
-
+    this._idle = false
+    this._idleObserver = null
     this.resetInvoked = 0
     this.releaseInvoked = 0
     this.seenQueries = []
@@ -101,6 +102,16 @@ export default class FakeConnection extends Connection {
     return this._idleTimestamp
   }
 
+  _setIdle (observer) {
+    this._idle = true
+    this._idleObserver = observer
+  }
+
+  _unsetIdle () {
+    this._idle = false
+    this._idleObserver = null
+  }
+
   protocol () {
     // return fake protocol object that simply records seen queries and parameters
     return {