Fix connection getting timeout while idle

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

@@ -54,6 +54,7 @@ export default class WebSocketChannel {
     this._receiveTimeout = null
     this._receiveTimeoutStarted = false
     this._receiveTimeoutId = null
+    this._closingPromise = null
 
     const { scheme, error } = determineWebSocketScheme(config, protocolSupplier)
     if (error) {
@@ -163,17 +164,23 @@ export default class WebSocketChannel {
    * @returns {Promise} A promise that will be resolved after channel is closed
    */
   close () {
-    return new Promise((resolve, reject) => {
-      this._clearConnectionTimeout()
-      if (this._ws && this._ws.readyState !== WS_CLOSED) {
-        this._open = false
-        this.stopReceiveTimeout()
-        this._ws.onclose = () => resolve()
-        this._ws.close()
-      } else {
-        resolve()
-      }
-    })
+    if (this._closingPromise === null) {
+      this._closingPromise = new Promise((resolve, reject) => {
+        this._clearConnectionTimeout()
+        if (this._ws && this._ws.readyState !== WS_CLOSED) {
+          this._open = false
+          this.stopReceiveTimeout()
+          this._ws.onclose = () => {
+            resolve()
+          }
+          this._ws.close()
+        } else {
+          resolve()
+        }
+      })
+    }
+
+    return this._closingPromise
   }
 
   /**

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,27 @@ 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()
+    return 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 +423,7 @@ export default class ChannelConnection extends Connection {
   }
 
   hasOngoingObservableRequests () {
-    return this._protocol.hasOngoingObservableRequests()
+    return !this._idle && this._protocol.hasOngoingObservableRequests()
   }
 
   /**
@@ -500,6 +521,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/channel/browser/browser-channel.test.js

@@ -323,6 +323,41 @@ describe('WebSocketChannel', () => {
         fakeSetTimeout.uninstall()
       }
     })
+
+    it('should return always the same promise', async () => {
+      const fakeSetTimeout = setTimeoutMock.install()
+      try {
+        // do not execute setTimeout callbacks
+        fakeSetTimeout.pause()
+        const address = ServerAddress.fromUrl('bolt://localhost:8989')
+        const driverConfig = { connectionTimeout: 4242 }
+        const channelConfig = new ChannelConfig(
+          address,
+          driverConfig,
+          SERVICE_UNAVAILABLE
+        )
+        webSocketChannel = new WebSocketChannel(
+          channelConfig,
+          undefined,
+          createWebSocketFactory(WS_OPEN)
+        )
+
+        const promise1 = webSocketChannel.close()
+        const promise2 = webSocketChannel.close()
+
+        expect(promise1).toBe(promise2)
+
+        await Promise.all([promise1, promise2])
+
+        const promise3 = webSocketChannel.close()
+
+        expect(promise3).toBe(promise2)
+
+        await promise3
+      } finally {
+        fakeSetTimeout.uninstall()
+      }
+    })
   })
 
   describe('.setupReceiveTimeout()', () => {

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/channel/browser/browser-channel.js

@@ -54,6 +54,7 @@ export default class WebSocketChannel {
     this._receiveTimeout = null
     this._receiveTimeoutStarted = false
     this._receiveTimeoutId = null
+    this._closingPromise = null
 
     const { scheme, error } = determineWebSocketScheme(config, protocolSupplier)
     if (error) {
@@ -163,17 +164,23 @@ export default class WebSocketChannel {
    * @returns {Promise} A promise that will be resolved after channel is closed
    */
   close () {
-    return new Promise((resolve, reject) => {
-      this._clearConnectionTimeout()
-      if (this._ws && this._ws.readyState !== WS_CLOSED) {
-        this._open = false
-        this.stopReceiveTimeout()
-        this._ws.onclose = () => resolve()
-        this._ws.close()
-      } else {
-        resolve()
-      }
-    })
+    if (this._closingPromise === null) {
+      this._closingPromise = new Promise((resolve, reject) => {
+        this._clearConnectionTimeout()
+        if (this._ws && this._ws.readyState !== WS_CLOSED) {
+          this._open = false
+          this.stopReceiveTimeout()
+          this._ws.onclose = () => {
+            resolve()
+          }
+          this._ws.close()
+        } else {
+          resolve()
+        }
+      })
+    }
+
+    return this._closingPromise
   }
 
   /**