Hide protocol object behind the Connection

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

@@ -82,7 +82,7 @@ export default class PooledConnectionProvider extends ConnectionProvider {
    */
   _createConnection ({ auth }, address, release) {
     return this._createChannelConnection(address).then(connection => {
-      connection._release = () => {
+      connection.release = () => {
         return release(address, connection)
       }
       this._openConnections[connection.id] = connection
@@ -160,7 +160,7 @@ export default class PooledConnectionProvider extends ConnectionProvider {
         await connection.resetAndFlush()
       }
     } finally {
-      await connection._release()
+      await connection.release()
     }
     return serverInfo
   }
@@ -191,7 +191,7 @@ export default class PooledConnectionProvider extends ConnectionProvider {
       }
       throw error
     } finally {
-      await Promise.all(connectionsToRelease.map(conn => conn._release()))
+      await Promise.all(connectionsToRelease.map(conn => conn.release()))
     }
   }
 
@@ -201,7 +201,7 @@ export default class PooledConnectionProvider extends ConnectionProvider {
     connection._sticky = connectionWithSameCredentials && !connection.supportsReAuth
 
     if (shouldCreateStickyConnection || connection._sticky) {
-      await connection._release()
+      await connection.release()
       throw newError('Driver is connected to a database that does not support user switch.')
     }
   }

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

@@ -156,6 +156,26 @@ export default class ChannelConnection extends Connection {
     }
   }
 
+  beginTransaction (config) {
+    return this._protocol.beginTransaction(config)
+  }
+
+  run (query, parameters, config) {
+    return this._protocol.run(query, parameters, config)
+  }
+
+  commitTransaction (config) {
+    return this._protocol.commitTransaction(config)
+  }
+
+  rollbackTransaction (config) {
+    return this._protocol.rollbackTransaction(config)
+  }
+
+  getProtocolVersion () {
+    return this._protocol.version
+  }
+
   get authToken () {
     return this._authToken
   }

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

@@ -35,6 +35,26 @@ export default class DelegateConnection extends Connection {
     this._delegate = delegate
   }
 
+  beginTransaction (config) {
+    return this._delegate.beginTransaction(config)
+  }
+
+  run (query, param, config) {
+    return this._delegate.run(query, param, config)
+  }
+
+  commitTransaction (config) {
+    return this._delegate.commitTransaction(config)
+  }
+
+  rollbackTransaction (config) {
+    return this._delegate.rollbackTransaction(config)
+  }
+
+  getProtocolVersion () {
+    return this._delegate.getProtocolVersion()
+  }
+
   get id () {
     return this._delegate.id
   }
@@ -103,11 +123,11 @@ export default class DelegateConnection extends Connection {
     return this._delegate.close()
   }
 
-  _release () {
+  release () {
     if (this._originalErrorHandler) {
       this._delegate._errorHandler = this._originalErrorHandler
     }
 
-    return this._delegate._release()
+    return this._delegate.release()
   }
 }

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

@@ -18,12 +18,14 @@
  */
 // eslint-disable-next-line no-unused-vars
 import { ResultStreamObserver, BoltProtocol } from '../bolt'
+import { Connection as CoreConnection } from 'neo4j-driver-core'
 
-export default class Connection {
+export default class Connection extends CoreConnection {
   /**
    * @param {ConnectionErrorHandler} errorHandler the error handler
    */
   constructor (errorHandler) {
+    super()
     this._errorHandler = errorHandler
   }
 
@@ -51,13 +53,6 @@ export default class Connection {
     throw new Error('not implemented')
   }
 
-  /**
-   * @returns {boolean} whether this connection is in a working condition
-   */
-  isOpen () {
-    throw new Error('not implemented')
-  }
-
   /**
    * @returns {BoltProtocol} the underlying bolt protocol assigned to this connection
    */
@@ -109,18 +104,6 @@ export default class Connection {
     throw new Error('not implemented')
   }
 
-  /**
-   * Send a RESET-message to the database. Message is immediately flushed to the network.
-   * @return {Promise<void>} promise resolved when SUCCESS-message response arrives, or failed when other response messages arrives.
-   */
-  resetAndFlush () {
-    throw new Error('not implemented')
-  }
-
-  hasOngoingObservableRequests () {
-    throw new Error('not implemented')
-  }
-
   /**
    * Call close on the channel.
    * @returns {Promise<void>} - A promise that will be resolved when the connection is closed.

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

@@ -273,53 +273,56 @@ class Pool {
   async _release (address, resource, pool) {
     const key = address.asKey()
 
-    if (pool.isActive()) {
-      // there exist idle connections for the given key
-      if (!await this._validateOnRelease(resource)) {
+    try {
+      if (pool.isActive()) {
+        // there exist idle connections for the given key
+        if (!await this._validateOnRelease(resource)) {
+          if (this._log.isDebugEnabled()) {
+            this._log.debug(
+              `${resource} destroyed and can't be released to the pool ${key} because it is not functional`
+            )
+          }
+          pool.removeInUse(resource)
+          await this._destroy(resource)
+        } else {
+          if (this._installIdleObserver) {
+            this._installIdleObserver(resource, {
+              onError: error => {
+                this._log.debug(
+                  `Idle connection ${resource} destroyed because of error: ${error}`
+                )
+                const pool = this._pools[key]
+                if (pool) {
+                  this._pools[key] = pool.filter(r => r !== resource)
+                  pool.removeInUse(resource)
+                }
+                // let's not care about background clean-ups due to errors but just trigger the destroy
+                // process for the resource, we especially catch any errors and ignore them to avoid
+                // unhandled promise rejection warnings
+                this._destroy(resource).catch(() => {})
+              }
+            })
+          }
+          pool.push(resource)
+          if (this._log.isDebugEnabled()) {
+            this._log.debug(`${resource} released to the pool ${key}`)
+          }
+        }
+      } else {
+        // key has been purged, don't put it back, just destroy the resource
         if (this._log.isDebugEnabled()) {
           this._log.debug(
-            `${resource} destroyed and can't be released to the pool ${key} because it is not functional`
+            `${resource} destroyed and can't be released to the pool ${key} because pool has been purged`
           )
         }
         pool.removeInUse(resource)
         await this._destroy(resource)
-      } else {
-        if (this._installIdleObserver) {
-          this._installIdleObserver(resource, {
-            onError: error => {
-              this._log.debug(
-                `Idle connection ${resource} destroyed because of error: ${error}`
-              )
-              const pool = this._pools[key]
-              if (pool) {
-                this._pools[key] = pool.filter(r => r !== resource)
-                pool.removeInUse(resource)
-              }
-              // let's not care about background clean-ups due to errors but just trigger the destroy
-              // process for the resource, we especially catch any errors and ignore them to avoid
-              // unhandled promise rejection warnings
-              this._destroy(resource).catch(() => {})
-            }
-          })
-        }
-        pool.push(resource)
-        if (this._log.isDebugEnabled()) {
-          this._log.debug(`${resource} released to the pool ${key}`)
-        }
       }
-    } else {
-      // key has been purged, don't put it back, just destroy the resource
-      if (this._log.isDebugEnabled()) {
-        this._log.debug(
-          `${resource} destroyed and can't be released to the pool ${key} because pool has been purged`
-        )
-      }
-      pool.removeInUse(resource)
-      await this._destroy(resource)
-    }
-    resourceReleased(key, this._activeResourceCounts)
+    } finally {
+      resourceReleased(key, this._activeResourceCounts)
 
-    this._processPendingAcquireRequests(address)
+      this._processPendingAcquireRequests(address)
+    }
   }
 
   async _purgeKey (key) {

packages/bolt-connection/test/connection-provider/connection-provider-direct.test.js

@@ -289,7 +289,7 @@ describe('constructor', () => {
 
         const connection = await create({}, server0, release)
 
-        const released = connection._release()
+        const released = connection.release()
 
         expect(released).toBe(releaseResult)
         expect(release).toHaveBeenCalledWith(server0, connection)
@@ -546,7 +546,7 @@ describe('user-switching', () => {
 
         expect(error).toEqual(newError('Driver is connected to a database that does not support user switch.'))
         expect(poolAcquire).toHaveBeenCalledWith({ auth: acquireAuth }, address)
-        expect(connection._release).toHaveBeenCalled()
+        expect(connection.release).toHaveBeenCalled()
         expect(connection._sticky).toEqual(isStickyConn)
       })
     })
@@ -599,15 +599,15 @@ describe('.verifyConnectivityAndGetServerInfo()', () => {
 
       await connectionProvider.verifyConnectivityAndGetServerInfo()
 
-      expect(seenConnections[0]._release).toHaveBeenCalledTimes(1)
+      expect(seenConnections[0].release).toHaveBeenCalledTimes(1)
     })
 
     it('should resetAndFlush and then release the connection', async () => {
       const { connectionProvider, seenConnections, resetAndFlush } = setup()
 
       await connectionProvider.verifyConnectivityAndGetServerInfo()
 
-      expect(seenConnections[0]._release.mock.invocationCallOrder[0])
+      expect(seenConnections[0].release.mock.invocationCallOrder[0])
         .toBeGreaterThan(resetAndFlush.mock.invocationCallOrder[0])
     })
 
@@ -636,7 +636,7 @@ describe('.verifyConnectivityAndGetServerInfo()', () => {
           await connectionProvider.verifyConnectivityAndGetServerInfo()
         } catch (e) {
         } finally {
-          expect(seenConnections[0]._release).toHaveBeenCalledTimes(1)
+          expect(seenConnections[0].release).toHaveBeenCalledTimes(1)
         }
       })
 
@@ -692,7 +692,7 @@ describe('.verifyConnectivityAndGetServerInfo()', () => {
         }
         connection.resetAndFlush = resetAndFlush
         if (releaseMock) {
-          connection._release = releaseMock
+          connection.release = releaseMock
         }
         seenConnections.push(connection)
         return connection
@@ -782,7 +782,7 @@ class FakeConnection extends Connection {
     super(null)
 
     this._address = address
-    this._release = jest.fn(() => release(address, this))
+    this.release = jest.fn(() => release(address, this))
     this._server = server
     this._authToken = auth
     this._closed = false