Don't send RESET message when the connection is not open

Send RESET when the connection is not optimal since it was time waiting for the message send times out. It could be really fast in the nodejs environment, but it could take some minutes in browser.

Skip send RESET when the connection is not open avoid this issue.

Author: bigmontz

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

@@ -270,7 +270,10 @@ export default class ChannelConnection extends Connection {
    */
   _handleFatalError (error) {
     this._isBroken = true
-    this._error = this.handleAndTransformError(this._protocol.currentFailure || error, this._address)
+    this._error = this.handleAndTransformError(
+      this._protocol.currentFailure || error,
+      this._address
+    )
 
     if (this._log.isErrorEnabled()) {
       this._log.error(
@@ -320,6 +323,10 @@ export default class ChannelConnection extends Connection {
   }
 
   _resetOnFailure () {
+    if (!this.isOpen()) {
+      return
+    }
+
     this._protocol.reset({
       onError: () => {
         this._protocol.resetFailure()

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

@@ -294,6 +294,38 @@ describe('WebSocketChannel', () => {
     }
   })
 
+  describe('.close()', () => {
+    it('should set _open to false before resolve the 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)
+        )
+
+        expect(webSocketChannel._open).toBe(true)
+
+        const promise = webSocketChannel.close()
+
+        expect(webSocketChannel._open).toBe(false)
+
+        await promise
+      } finally {
+        fakeSetTimeout.uninstall()
+      }
+    })
+  })
+
   describe('.setupReceiveTimeout()', () => {
     beforeEach(() => {
       const address = ServerAddress.fromUrl('http://localhost:8989')

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

@@ -44,6 +44,20 @@ describe('NodeChannel', () => {
     return expect(channel.close()).resolves.not.toThrow()
   })
 
+  describe('.close()', () => {
+    it('should set _open to false before resolve the promise', async () => {
+      const channel = createMockedChannel(true)
+
+      expect(channel._open).toBe(true)
+
+      const promise = channel.close()
+
+      expect(channel._open).toBe(false)
+
+      await promise
+    })
+  })
+
   describe('.setupReceiveTimeout()', () => {
     it('should call socket.setTimeout(receiveTimeout)', () => {
       const receiveTimeout = 42

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

@@ -19,7 +19,6 @@
 
 import ChannelConnection from '../../src/connection/connection-channel'
 import { int, internal } from 'neo4j-driver-core'
-import { add } from 'lodash'
 
 const {
   serverAddress: { ServerAddress },
@@ -127,6 +126,69 @@ describe('ChannelConnection', () => {
     )
   })
 
+  describe('._resetOnFailure()', () => {
+    describe('when connection isOpen', () => {
+      it('should call protocol.reset() and then protocol.resetFailure() onComplete', () => {
+        const channel = {
+          _open: true
+        }
+
+        const protocol = {
+          reset: jest.fn(observer => observer.onComplete()),
+          resetFailure: jest.fn()
+        }
+        const protocolSupplier = () => protocol
+        const connection = spyOnConnectionChannel({ channel, protocolSupplier })
+
+        connection._resetOnFailure()
+
+        expect(protocol.reset).toHaveBeenCalled()
+        expect(protocol.resetFailure).toHaveBeenCalled()
+      })
+
+      it('should call protocol.reset() and then protocol.resetFailure() onError', () => {
+        const channel = {
+          _open: true
+        }
+
+        const protocol = {
+          reset: jest.fn(observer => observer.onError()),
+          resetFailure: jest.fn()
+        }
+        const protocolSupplier = () => protocol
+        const connection = spyOnConnectionChannel({ channel, protocolSupplier })
+
+        connection._resetOnFailure()
+
+        expect(protocol.reset).toHaveBeenCalled()
+        expect(protocol.resetFailure).toHaveBeenCalled()
+      })
+    })
+
+    describe('when connection is not open', () => {
+      it('should not call protocol.reset() and protocol.resetFailure()', () => {
+        const channel = {
+          _open: false
+        }
+
+        const protocol = {
+          reset: jest.fn(observer => {
+            observer.onComplete()
+            observer.onError()
+          }),
+          resetFailure: jest.fn()
+        }
+        const protocolSupplier = () => protocol
+        const connection = spyOnConnectionChannel({ channel, protocolSupplier })
+
+        connection._resetOnFailure()
+
+        expect(protocol.reset).not.toHaveBeenCalled()
+        expect(protocol.resetFailure).not.toHaveBeenCalled()
+      })
+    })
+  })
+
   function spyOnConnectionChannel ({
     channel,
     errorHandler,

core/src/internal/connection-holder.ts

@@ -179,10 +179,13 @@ class ConnectionHolder implements ConnectionHolderInterface {
     this._connectionPromise = this._connectionPromise
       .then((connection?: Connection) => {
         if (connection) {
-          return connection
-            .resetAndFlush()
-            .catch(ignoreError)
-            .then(() => connection._release())
+          if (connection.isOpen()) {
+            return connection
+              .resetAndFlush()
+              .catch(ignoreError)
+              .then(() => connection._release())
+          }
+          return connection._release()
         } else {
           return Promise.resolve()
         }

test/internal/connection-holder.test.js

@@ -292,6 +292,116 @@ describe('#unit ConnectionHolder', () => {
 
     expect(connectionHolder.database()).toBe('testdb')
   })
+
+  describe('.releaseConnection()', () => {
+    describe('when the connection is initialized', () => {
+      describe('and connection is open', () => {
+        let connection
+
+        beforeEach(async () => {
+          connection = new FakeConnection()
+          const connectionProvider = newSingleConnectionProvider(connection)
+          const connectionHolder = new ConnectionHolder({
+            mode: READ,
+            connectionProvider
+          })
+
+          connectionHolder.initializeConnection()
+
+          await connectionHolder.releaseConnection()
+        })
+
+        it('should call connection.resetAndFlush', () => {
+          expect(connection.resetInvoked).toBe(1)
+        })
+
+        it('should call connection._release()', () => {
+          expect(connection.releaseInvoked).toBe(1)
+        })
+      })
+
+      describe('and connection is not open', () => {
+        let connection
+
+        beforeEach(async () => {
+          connection = new FakeConnection()
+          connection._open = false
+          const connectionProvider = newSingleConnectionProvider(connection)
+          const connectionHolder = new ConnectionHolder({
+            mode: READ,
+            connectionProvider
+          })
+
+          connectionHolder.initializeConnection()
+
+          await connectionHolder.releaseConnection()
+        })
+
+        it('should call connection.resetAndFlush', () => {
+          expect(connection.resetInvoked).toBe(0)
+        })
+
+        it('should call connection._release()', () => {
+          expect(connection.releaseInvoked).toBe(1)
+        })
+      })
+    })
+  })
+
+  describe('.close()', () => {
+    describe('when the connection is initialized', () => {
+      describe('and connection is open', () => {
+        let connection
+
+        beforeEach(async () => {
+          connection = new FakeConnection()
+          const connectionProvider = newSingleConnectionProvider(connection)
+          const connectionHolder = new ConnectionHolder({
+            mode: READ,
+            connectionProvider
+          })
+
+          connectionHolder.initializeConnection()
+
+          await connectionHolder.close()
+        })
+
+        it('should call connection.resetAndFlush', () => {
+          expect(connection.resetInvoked).toBe(1)
+        })
+
+        it('should call connection._release()', () => {
+          expect(connection.releaseInvoked).toBe(1)
+        })
+      })
+
+      describe('and connection is not open', () => {
+        let connection
+
+        beforeEach(async () => {
+          connection = new FakeConnection()
+          connection._open = false
+          const connectionProvider = newSingleConnectionProvider(connection)
+          const connectionHolder = new ConnectionHolder({
+            mode: READ,
+            connectionProvider
+          })
+
+          connectionHolder.initializeConnection()
+
+          await connectionHolder.close()
+        })
+
+        it('should call connection.resetAndFlush', () => {
+          expect(connection.resetInvoked).toBe(0)
+        })
+
+        it('should call connection._release()', () => {
+          expect(connection.releaseInvoked).toBe(1)
+        })
+      })
+    })
+  })
 })
 
 class RecordingConnectionProvider extends SingleConnectionProvider {

test/result.test.js

@@ -73,7 +73,7 @@ describe('#integration result stream', () => {
         done()
       },
       onError: error => {
-        console.log(error)
+        done.fail(error)
       }
     })
   })