Hide protocol object behind the Connection

The connection in the driver is exposing the protocol object, which is not great. Improve this part of the code can be done by make the connection object have methods to do high level requests to the server.

List of calls the core driver does using the protocol, by passing the connection:

* `version`
* `run`
* `beginTransaction`
* `commitTransaction`
* `rollbackTransaction`

Methods is present in the `Connection` interface (used by core in **bold**):

* `id`
* `database`
* `server`
* `authToken`
* `address`
* `version`
* `supportsReAuth`
* **`isOpen`**
* **`protocol`**
* `connect`
* `write`
* **`resetAndFlush`**
* **`hasOngoingObservableRequests`**
* **`_release`**

So, `isOpen`, `resetAndFlush` and `hasOngoingObservableRequests`  are the methods which will stay in the connection along with the new methods. The method `release` will move the a `Releasable` interface, which will be composed with `Connection` when returning the connection from the provider.

The `Releasable` interface is also defined to enable the `ConnectionProvider` returns a connection which can be released back to the pool.

Internally, `bolt-connection` can keep exposing the internal of the connection outside the connection in a first moment. The full encapsulation of the `protocol` should be done in the next phase of refactoring.

Author: bigmontz

packages/core/src/connection-provider.ts

@@ -23,6 +23,18 @@ import { bookmarks } from './internal'
 import { ServerInfo } from './result-summary'
 import { AuthToken } from './types'
 
+/**
+ * Interface define a releasable resource shape
+ *
+ * @private
+ * @interface
+ */
+class Releasable {
+  release (): Promise<void> {
+    throw new Error('Not implemented')
+  }
+}
+
 /**
  * Interface define a common way to acquire a connection
  *
@@ -53,7 +65,7 @@ class ConnectionProvider {
     impersonatedUser?: string
     onDatabaseNameResolved?: (databaseName?: string) => void
     auth?: AuthToken
-  }): Promise<Connection> {
+  }): Promise<Connection & Releasable> {
     throw Error('Not implemented')
   }
 
@@ -150,3 +162,6 @@ class ConnectionProvider {
 }
 
 export default ConnectionProvider
+export {
+  Releasable
+}

packages/core/src/connection.ts

@@ -18,121 +18,91 @@
  */
 /* eslint-disable @typescript-eslint/promise-function-async */
 
-import { ServerAddress } from './internal/server-address'
-
-/**
- * Interface which defines the raw connection with the database
- * @private
- */
-class Connection {
-  get id (): string {
-    return ''
-  }
+import { Bookmarks } from './internal/bookmarks'
+import { AccessMode } from './internal/constants'
+import { ResultStreamObserver } from './internal/observers'
+import { TxConfig } from './internal/tx-config'
+import NotificationFilter from './notification-filter'
+
+interface HasBeforeErrorAndAfterComplete {
+  beforeError?: (error: Error) => void
+  afterComplete?: (metadata: unknown) => void
+}
 
-  get databaseId (): string {
-    return ''
-  }
+interface BeginTransactionConfig extends HasBeforeErrorAndAfterComplete {
+  bookmarks: Bookmarks
+  txConfig: TxConfig
+  mode?: AccessMode
+  database?: string
+  impersonatedUser?: string
+  notificationFilter?: NotificationFilter
+}
 
-  get server (): any {
-    return {}
-  }
+interface CommitTransactionConfig extends HasBeforeErrorAndAfterComplete {
 
-  /**
-   * @property {object} authToken The auth registered in the connection
-   */
-  get authToken (): any {
-    return {}
-  }
+}
 
-  /**
-   * @property {ServerAddress} the server address this connection is opened against
-   */
-  get address (): ServerAddress | undefined {
-    return undefined
-  }
+interface RollbackConnectionConfig extends HasBeforeErrorAndAfterComplete {
 
-  /**
-   * @property {ServerVersion} the version of the server this connection is connected to
-   */
-  get version (): any {
-    return undefined
-  }
+}
 
-  /**
-   * @property {boolean} supportsReAuth Indicates the connection supports re-auth
-   */
-  get supportsReAuth (): boolean {
-    return false
-  }
+interface RunQueryConfig extends BeginTransactionConfig {
+  fetchSize: number
+  highRecordWatermark: number
+  lowRecordWatermark: number
+  reactive: boolean
+}
 
-  /**
-   * @returns {boolean} whether this connection is in a working condition
-   */
-  isOpen (): boolean {
-    return false
+/**
+ * Interface which defines a connection for the core driver object.
+ *
+ *
+ * This connection exposes only methods used by the code module.
+ * Methods with connection implementation details can be defined and used
+ * by the implementation layer.
+ *
+ * @private
+ * @interface
+ */
+class Connection {
+  beginTransaction (config: BeginTransactionConfig): ResultStreamObserver {
+    throw new Error('Not implemented')
   }
 
-  /**
-   * @todo be removed and internalize the methods
-   * @returns {any} the underlying bolt protocol assigned to this connection
-   */
-  protocol (): any {
-    throw Error('Not implemented')
+  run (query: string, parameters?: Record<string, unknown>, config?: RunQueryConfig): ResultStreamObserver {
+    throw new Error('Not implemented')
   }
 
-  /**
-   * Connect to the target address, negotiate Bolt protocol and send initialization message.
-   * @param {string} userAgent the user agent for this driver.
-   * @param {string} boltAgent the bolt agent for this driver.
-   * @param {Object} authToken the object containing auth information.
-   * @param {Object} waitReAuth whether to connect method should wait until re-Authorised
-   * @return {Promise<Connection>} promise resolved with the current connection if connection is successful. Rejected promise otherwise.
-   */
-  connect (userAgent: string, boltAgent: string, authToken: any, waitReAuth: false): Promise<Connection> {
-    throw Error('Not implemented')
+  commitTransaction (config: CommitTransactionConfig): ResultStreamObserver {
+    throw new Error('Not implemented')
   }
 
-  /**
-   * Write a message to the network channel.
-   * @param {RequestMessage} message the message to write.
-   * @param {ResultStreamObserver} observer the response observer.
-   * @param {boolean} flush `true` if flush should happen after the message is written to the buffer.
-   */
-  write (message: any, observer: any, flush: boolean): void {
-    throw Error('Not implemented')
+  rollbackTransaction (config: RollbackConnectionConfig): ResultStreamObserver {
+    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 (): Promise<void> {
-    throw Error('Not implemented')
+    throw new Error('Not implemented')
   }
 
-  /**
-   * Checks if there is an ongoing request being handled
-   * @return {boolean} `true` if there is an ongoing request being handled
-   */
-  hasOngoingObservableRequests (): boolean {
-    throw Error('Not implemented')
+  isOpen (): boolean {
+    throw new Error('Not implemented')
   }
 
-  /**
-   * Call close on the channel.
-   * @returns {Promise<void>} - A promise that will be resolved when the connection is closed.
-   *
-   */
-  close (): Promise<void> {
-    throw Error('Not implemented')
+  getProtocolVersion (): number {
+    throw new Error('Not implemented')
   }
 
-  /**
-   * Called to release the connection
-   */
-  _release (): Promise<void> {
-    return Promise.resolve()
+  hasOngoingObservableRequests (): boolean {
+    throw new Error('Not implemented')
   }
 }
 
 export default Connection
+
+export type {
+  BeginTransactionConfig,
+  CommitTransactionConfig,
+  RollbackConnectionConfig,
+  RunQueryConfig
+}

packages/core/src/internal/connection-holder.ts

@@ -21,9 +21,9 @@
 import { newError } from '../error'
 import { assertString } from './util'
 import Connection from '../connection'
-import { ACCESS_MODE_WRITE } from './constants'
+import { ACCESS_MODE_WRITE, AccessMode } from './constants'
 import { Bookmarks } from './bookmarks'
-import ConnectionProvider from '../connection-provider'
+import ConnectionProvider, { Releasable } from '../connection-provider'
 import { AuthToken } from '../types'
 
 /**
@@ -77,12 +77,12 @@ interface ConnectionHolderInterface {
  * @private
  */
 class ConnectionHolder implements ConnectionHolderInterface {
-  private readonly _mode: string
+  private readonly _mode: AccessMode
   private _database?: string
   private readonly _bookmarks: Bookmarks
   private readonly _connectionProvider?: ConnectionProvider
   private _referenceCount: number
-  private _connectionPromise: Promise<Connection | null>
+  private _connectionPromise: Promise<Connection & Releasable | null>
   private readonly _impersonatedUser?: string
   private readonly _getConnectionAcquistionBookmarks: () => Promise<Bookmarks>
   private readonly _onDatabaseNameResolved?: (databaseName?: string) => void
@@ -111,7 +111,7 @@ class ConnectionHolder implements ConnectionHolderInterface {
     getConnectionAcquistionBookmarks,
     auth
   }: {
-    mode?: string
+    mode?: AccessMode
     database?: string
     bookmarks?: Bookmarks
     connectionProvider?: ConnectionProvider
@@ -133,7 +133,7 @@ class ConnectionHolder implements ConnectionHolderInterface {
     this._getConnectionAcquistionBookmarks = getConnectionAcquistionBookmarks ?? (() => Promise.resolve(Bookmarks.empty()))
   }
 
-  mode (): string | undefined {
+  mode (): AccessMode | undefined {
     return this._mode
   }
 
@@ -168,7 +168,7 @@ class ConnectionHolder implements ConnectionHolderInterface {
     return true
   }
 
-  private async _createConnectionPromise (connectionProvider: ConnectionProvider): Promise<Connection | null> {
+  private async _createConnectionPromise (connectionProvider: ConnectionProvider): Promise<Connection & Releasable | null> {
     return await connectionProvider.acquireConnection({
       accessMode: this._mode,
       database: this._database,
@@ -218,15 +218,15 @@ class ConnectionHolder implements ConnectionHolderInterface {
    */
   private _releaseConnection (hasTx?: boolean): Promise<Connection | null> {
     this._connectionPromise = this._connectionPromise
-      .then((connection?: Connection | null) => {
+      .then((connection?: Connection & Releasable | null) => {
         if (connection != null) {
           if (connection.isOpen() && (connection.hasOngoingObservableRequests() || hasTx === true)) {
             return connection
               .resetAndFlush()
               .catch(ignoreError)
-              .then(() => connection._release().then(() => null))
+              .then(() => connection.release().then(() => null))
           }
-          return connection._release().then(() => null)
+          return connection.release().then(() => null)
         } else {
           return Promise.resolve(null)
         }

packages/core/src/internal/constants.ts

@@ -38,6 +38,8 @@ const BOLT_PROTOCOL_V5_1: number = 5.1
 const BOLT_PROTOCOL_V5_2: number = 5.2
 const BOLT_PROTOCOL_V5_3: number = 5.3
 
+export type AccessMode = typeof ACCESS_MODE_READ | typeof ACCESS_MODE_WRITE
+
 export {
   FETCH_ALL,
   ACCESS_MODE_READ,

packages/core/src/result.ts

@@ -513,7 +513,7 @@ class Result<R extends RecordShape = RecordShape> implements Promise<QueryResult
           connectionHolder
             .releaseConnection()
             .then(() =>
-              connection?.protocol()?.version
+              connection?.getProtocolVersion()
             ),
         // onRejected:
         _ => undefined

packages/core/src/session.ts

@@ -187,7 +187,7 @@ class Session {
     const result = this._run(validatedQuery, params, async connection => {
       const bookmarks = await this._bookmarks()
       this._assertSessionIsOpen()
-      return (connection as Connection).protocol().run(validatedQuery, params, {
+      return (connection as Connection).run(validatedQuery, params, {
         bookmarks,
         txConfig: autoCommitTxConfig,
         mode: this._mode,

packages/core/src/transaction.ts

@@ -139,7 +139,7 @@ class Transaction {
         this._onConnection()
         if (connection != null) {
           this._bookmarks = await getBookmarks()
-          return connection.protocol().beginTransaction({
+          return connection.beginTransaction({
             bookmarks: this._bookmarks,
             txConfig,
             mode: this._connectionHolder.mode(),
@@ -150,13 +150,13 @@ class Transaction {
               if (events != null) {
                 events.onError(error)
               }
-              return this._onError(error)
+              this._onError(error).catch(() => {})
             },
             afterComplete: (metadata: any) => {
               if (events != null) {
                 events.onComplete(metadata)
               }
-              return this._onComplete(metadata)
+              this._onComplete(metadata)
             }
           })
         } else {
@@ -364,7 +364,7 @@ const _states = {
       }
     },
     run: (
-      query: Query,
+      query: string,
       parameters: any,
       {
         connectionHolder,
@@ -388,7 +388,7 @@ const _states = {
           .then(conn => {
             onConnection()
             if (conn != null) {
-              return conn.protocol().run(query, parameters, {
+              return conn.run(query, parameters, {
                 bookmarks: Bookmarks.empty(),
                 txConfig: TxConfig.empty(),
                 beforeError: onError,
@@ -643,12 +643,12 @@ function finishTransaction (
         return Promise.all(pendingResults.map(result => result.summary())).then(results => {
           if (connection != null) {
             if (commit) {
-              return connection.protocol().commitTransaction({
+              return connection.commitTransaction({
                 beforeError: onError,
                 afterComplete: onComplete
               })
             } else {
-              return connection.protocol().rollbackTransaction({
+              return connection.rollbackTransaction({
                 beforeError: onError,
                 afterComplete: onComplete
               })