Introduce the hability of awaiting for eager begining a tx

Author: bigmontz

packages/core/src/session.ts

@@ -31,6 +31,7 @@ import ConnectionProvider from './connection-provider'
 import { Query, SessionMode } from './types'
 import Connection from './connection'
 import { NumberOrInteger } from './graph-types'
+import TransactionPromise from './transaction-promise'
 
 type ConnectionConsumer = (connection: Connection | void) => any | undefined
 type TransactionWork<T> = (tx: Transaction) => Promise<T> | T
@@ -241,7 +242,7 @@ class Session {
    * @param {TransactionConfig} [transactionConfig] - Configuration for the new auto-commit transaction.
    * @returns {Transaction} New Transaction.
    */
-  beginTransaction(transactionConfig?: TransactionConfig): Transaction {
+  beginTransaction(transactionConfig?: TransactionConfig): TransactionPromise {
     // this function needs to support bookmarks parameter for backwards compatibility
     // parameter was of type {string|string[]} and represented either a single or multiple bookmarks
     // that's why we need to check parameter type and decide how to interpret the value
@@ -255,7 +256,7 @@ class Session {
     return this._beginTransaction(this._mode, txConfig)
   }
 
-  _beginTransaction(accessMode: SessionMode, txConfig: TxConfig): Transaction {
+  _beginTransaction(accessMode: SessionMode, txConfig: TxConfig): TransactionPromise {
     if (!this._open) {
       throw newError('Cannot begin a transaction on a closed session.')
     }
@@ -271,7 +272,7 @@ class Session {
     connectionHolder.initializeConnection()
     this._hasTx = true
 
-    const tx = new Transaction({
+    const tx = new TransactionPromise({
       connectionHolder,
       impersonatedUser: this._impersonatedUser,
       onClose: this._transactionClosed.bind(this),

packages/core/src/transaction-promise.ts

@@ -0,0 +1,162 @@
+/**
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Transaction from "./transaction"
+import {
+  ConnectionHolder
+} from './internal/connection-holder'
+
+import { Bookmarks } from './internal/bookmarks'
+import { TxConfig } from "./internal/tx-config";
+import Connection from "./connection";
+import Result from "./result";
+import { Query } from "./types";
+
+class TransactionPromise extends Transaction implements Promise<Transaction>{
+  private _beginError?: Error;
+  private _beginMetadata?: any;
+  private _beginPromise?: Promise<Transaction>;
+  private _reject?: (error: Error) => void;
+  private _resolve?: (value?: Transaction | PromiseLike<Transaction> | undefined) => void;
+  private _delegate: Transaction;
+
+  constructor({
+    connectionHolder,
+    onClose,
+    onBookmarks,
+    onConnection,
+    reactive,
+    fetchSize,
+    impersonatedUser,
+    highRecordWatermark,
+    lowRecordWatermark
+  }: {
+    connectionHolder: ConnectionHolder
+    onClose: () => void
+    onBookmarks: (bookmarks: Bookmarks) => void
+    onConnection: () => void
+    reactive: boolean
+    fetchSize: number
+    impersonatedUser?: string,
+    highRecordWatermark: number,
+    lowRecordWatermark: number
+  }) {
+    super({
+      connectionHolder,
+      onClose,
+      onBookmarks,
+      onConnection,
+      reactive,
+      fetchSize,
+      impersonatedUser,
+      highRecordWatermark,
+      lowRecordWatermark
+    })
+    this._delegate = new Transaction({
+      connectionHolder,
+      onClose,
+      onBookmarks,
+      onConnection,
+      reactive,
+      fetchSize,
+      impersonatedUser,
+      highRecordWatermark,
+      lowRecordWatermark
+    })
+  }
+  run(query: Query, parameters?: any): Result {
+    return this._delegate.run(query, parameters);
+  }
+  commit(): Promise<void> {
+    return this._delegate.commit()
+  }
+  rollback(): Promise<void> {
+    return this._delegate.rollback();
+  }
+  isOpen(): boolean {
+    return this._delegate.isOpen()
+  }
+  close(): Promise<void> {
+    return this._delegate.close();
+  }
+  _onErrorCallback(err: any): Promise<void | Connection> {
+    return this._delegate._onErrorCallback(err);
+  }
+  _onCompleteCallback(meta: { bookmark?: string | string[] | undefined; }): void {
+    return this._delegate._onCompleteCallback(meta);
+  }
+
+  _begin(bookmarks: string | Bookmarks | string[], txConfig: TxConfig): void {
+    return this._delegate._begin(bookmarks, txConfig, {
+      onError: this._onBeginError.bind(this),
+      onComplete: this._onBeginMetadata.bind(this)
+    });
+  }
+
+  _onBeginError(error: Error): void {
+    this._beginError = error;
+    if (this._reject) {
+      this._reject(error);
+    }
+  }
+
+  _onBeginMetadata(metadata: any): void {
+    this._beginMetadata = metadata || {};
+    if (this._resolve) {
+      this._resolve(this._delegate);
+    }
+  }
+
+  [Symbol.toStringTag]: string = "TransactionPromise"
+
+  then<TResult1 = Transaction, TResult2 = never>(
+    onfulfilled?: 
+      ((value: Transaction) => TResult1 | PromiseLike<TResult1> )
+      | null, 
+    onrejected?: 
+      ((reason: any) => TResult2 | PromiseLike<TResult2>)
+      | null
+    ) : Promise<TResult1 | TResult2> {
+    return this._getOrCreateBeginPromise().then(onfulfilled, onrejected);
+  }
+  catch<TResult = never>(onrejected?: ((reason: any) => TResult | PromiseLike<TResult>) | null): Promise<any> {
+    return this._getOrCreateBeginPromise().catch(onrejected);
+  }
+  finally(onfinally?: (() => void) | null): Promise<Transaction> {
+    return this._getOrCreateBeginPromise().finally(onfinally);
+  }
+
+  private _getOrCreateBeginPromise(): Promise<Transaction> {
+    if (!this._beginPromise) {
+      this._beginPromise = new Promise((resolve, reject) => {
+        this._resolve = resolve;
+        this._reject = reject;
+        if (this._beginError) {
+          reject(this._beginError);
+        }
+        if (this._beginMetadata) {
+          resolve(this._delegate);
+        }
+      });
+    } 
+    return this._beginPromise;
+  }
+}
+
+export default TransactionPromise

packages/core/src/transaction.ts

@@ -109,7 +109,10 @@ class Transaction {
    * @param {TxConfig} txConfig
    * @returns {void}
    */
-  _begin(bookmarks: Bookmarks | string | string[], txConfig: TxConfig): void {
+  _begin(bookmarks: Bookmarks | string | string[], txConfig: TxConfig, events?: {
+    onError: (error: Error) => void
+    onComplete: (metadata: any) => void
+  }): void {
     this._connectionHolder
       .getConnection()
       .then(connection => {
@@ -121,14 +124,29 @@ class Transaction {
             mode: this._connectionHolder.mode(),
             database: this._connectionHolder.database(),
             impersonatedUser: this._impersonatedUser,
-            beforeError: this._onError,
-            afterComplete: this._onComplete
+            beforeError: (error: Error) => {
+              if (events) {
+                events.onError(error)
+              }
+              return this._onError(error).catch(() => {})
+            },
+            afterComplete: (metadata: any) => {
+              if (events) {
+                events.onComplete(metadata)
+              }
+              return this._onComplete(metadata)
+            }
           })
         } else {
           throw newError('No connection available')
         }
       })
-      .catch(error => this._onError(error))
+      .catch(error => { 
+        if (events) {
+          events.onError(error)
+        }
+        this._onError(error).catch(() => {})
+      })
   }
 
   /**

packages/testkit-backend/src/request-handlers.js

@@ -285,16 +285,21 @@ export function RetryableNegative (context, data, wire) {
 export function SessionBeginTransaction (context, data, wire) {
   const { sessionId, txMeta: metadata, timeout } = data
   const session = context.getSession(sessionId)
-  let tx
+  
   try {
-    tx = session.beginTransaction({ metadata, timeout })
+    return session.beginTransaction({ metadata, timeout })
+    .then(tx => {
+      const id = context.addTx(tx, sessionId)
+      wire.writeResponse('Transaction', { id })
+    }).catch(e => {
+      console.log('got some err: ' + JSON.stringify(e))
+      wire.writeError(e)
+    })
   } catch (e) {
     console.log('got some err: ' + JSON.stringify(e))
     wire.writeError(e)
     return
   }
-  const id = context.addTx(tx, sessionId)
-  wire.writeResponse('Transaction', { id })
 }
 
 export function TransactionCommit (context, data, wire) {
@@ -372,6 +377,7 @@ export function GetFeatures (_context, _params, wire) {
       'Feature:Bolt:4.4',
       'Feature:API:Result.List',
       'Feature:API:Result.Peek',
+      'Optimization:EagerTransactionBegin',
       'Temporary:ConnectionAcquisitionTimeout',
       'Temporary:CypherPathAndRelationship',
       'Temporary:DriverFetchSize',

packages/testkit-backend/src/skipped-tests/common.js

@@ -1,6 +1,11 @@
 import skip, { ifEquals, ifEndsWith, ifStartsWith } from './skip'
 
 const skippedTests = [
+  skip(
+    'Eager verification not implemented for tx functions',
+    ifEquals('stub.tx_run.test_tx_run.TestTxRun.test_eager_begin_on_tx_func_run_with_error_on_begin'),
+    ifEquals('stub.tx_run.test_tx_run.TestTxRun.test_eager_begin_on_tx_func_run_with_disconnect_on_begin')
+  ),
   skip(
     'Fail while enable Temporary::ResultKeys',
     ifEquals('neo4j.test_bookmarks.TestBookmarks.test_can_pass_bookmark_into_next_session'),