Add executeRead and executeWrite to RxSession

Author: bigmontz

packages/neo4j-driver/src/session-rx.js

@@ -84,6 +84,7 @@ export default class RxSession {
    * Executes the provided unit of work in a {@link READ} reactive transaction which is created with the provided
    * transaction configuration.
    * @public
+   * @deprecated This method will be removed in version 6.0. Please, use {@link RxSession#executeRead} instead.
    * @param {function(txc: RxTransaction): Observable} work - A unit of work to be executed.
    * @param {TransactionConfig} transactionConfig - Configuration for the enclosing transaction created by the driver.
    * @returns {Observable} - A reactive stream returned by the unit of work.
@@ -96,6 +97,7 @@ export default class RxSession {
    * Executes the provided unit of work in a {@link WRITE} reactive transaction which is created with the provided
    * transaction configuration.
    * @public
+   * @deprecated This method will be removed in version 6.0. Please, use {@link RxSession#executeWrite} instead.
    * @param {function(txc: RxTransaction): Observable} work - A unit of work to be executed.
    * @param {TransactionConfig} transactionConfig - Configuration for the enclosing transaction created by the driver.
    * @returns {Observable} - A reactive stream returned by the unit of work.
@@ -104,6 +106,45 @@ export default class RxSession {
     return this._runTransaction(ACCESS_MODE_WRITE, work, transactionConfig)
   }
 
+
+  /**
+   * Executes the provided unit of work in a {@link READ} reactive transaction which is created with the provided
+   * transaction configuration.
+   * @public
+   * @param {function(txc: RxManagedTransaction): Observable} work - A unit of work to be executed.
+   * @param {TransactionConfig} transactionConfig - Configuration for the enclosing transaction created by the driver.
+   * @returns {Observable} - A reactive stream returned by the unit of work.
+   */
+   executeRead (work, transactionConfig) {
+    return this._executeInTransaction(ACCESS_MODE_READ, work, transactionConfig)
+  }
+
+  /**
+   * Executes the provided unit of work in a {@link WRITE} reactive transaction which is created with the provided
+   * transaction configuration.
+   * @public
+   * @param {function(txc: RxManagedTransaction): Observable} work - A unit of work to be executed.
+   * @param {TransactionConfig} transactionConfig - Configuration for the enclosing transaction created by the driver.
+   * @returns {Observable} - A reactive stream returned by the unit of work.
+   */
+  executeWrite (work, transactionConfig) {
+    return this._executeInTransaction(ACCESS_MODE_WRITE, work, transactionConfig)
+  }
+
+  /**
+   * @private
+   * @param {function(txc: RxManagedTransaction): Observable} work
+   * @param {TransactionConfig} transactionConfig
+   * @returns {Observable}
+   */
+  _executeInTransaction (accessMode, work, transactionConfig) {
+    const wrapper = txc => new RxManagedTransaction({
+      run: txc.run.bind(txc),
+      isOpen: txc.isOpen.bind(txc)
+    })
+    return this._runTransaction(accessMode, work, transactionConfig, wrapper)
+  }
+
   /**
    * Closes this reactive session.
    *
@@ -181,7 +222,7 @@ export default class RxSession {
   /**
    * @private
    */
-  _runTransaction (accessMode, work, transactionConfig) {
+  _runTransaction (accessMode, work, transactionConfig, transactionWrapper = (tx) => tx) {
     let txConfig = TxConfig.empty()
     if (transactionConfig) {
       txConfig = new TxConfig(transactionConfig)
@@ -192,7 +233,7 @@ export default class RxSession {
         flatMap(txc =>
           defer(() => {
             try {
-              return work(txc)
+              return work(transactionWrapper(txc))
             } catch (err) {
               return throwError(err)
             }

packages/neo4j-driver/src/transaction-managed-rx.js

@@ -0,0 +1,50 @@
+/**
+ * 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.
+ */
+
+/**
+ * Represents a rx transaction that is managed by the transaction executor.
+ * 
+ * @public
+ */
+class RxManagedTransaction {
+  constructor({ run, isOpen }) {
+    this._run = run
+    this._isOpen = isOpen
+  }
+
+  /**
+   * Creates a reactive result that will execute the query in this transaction, with the provided parameters.
+   *
+   * @public
+   * @param {string} query - Query to be executed.
+   * @param {Object} parameters - Parameter values to use in query execution.
+   * @returns {RxResult} - A reactive result
+   */
+   run (query, parameters) {
+     return this._run(query, parameters)
+   }
+
+   /**
+   * Check if this transaction is active, which means commit and rollback did not happen.
+   * @return {boolean} `true` when not committed and not rolled back, `false` otherwise.
+   */
+    isOpen() {
+      return this._isOpen()
+    }
+}

packages/neo4j-driver/src/transaction-rx.js

@@ -41,7 +41,6 @@ export default class RxTransaction {
    * @param {Object} parameters - Parameter values to use in query execution.
    * @returns {RxResult} - A reactive result
    */
-
   run (query, parameters) {
     return new RxResult(
       new Observable(observer => {
@@ -91,6 +90,14 @@ export default class RxTransaction {
     })
   }
 
+  /**
+   * Check if this transaction is active, which means commit and rollback did not happen.
+   * @return {boolean} `true` when not committed and not rolled back, `false` otherwise.
+   */
+   isOpen() {
+    return this._txc.isOpen()
+  }
+
   /**
    * Closes the transaction
    *

packages/neo4j-driver/test/rx/session.test.js

@@ -239,6 +239,138 @@ describe('#integration rx-session', () => {
     expect(await countNodes('Person')).toBe(0)
   }, 60000)
 
+  describe('.executeWrite()', () => {
+    it('should run transactions without retries', async () => {
+      if (protocolVersion < 4.0) {
+        return
+      }
+  
+      const txcWork = new ConfigurableTransactionWork({
+        query: 'CREATE (:WithoutRetry) RETURN 5'
+      })
+  
+      const result = await session
+        .executeWrite(txc => txcWork.work(txc))
+        .pipe(materialize(), toArray())
+        .toPromise()
+      expect(result).toEqual([
+        Notification.createNext(5),
+        Notification.createComplete()
+      ])
+  
+      expect(txcWork.invocations).toBe(1)
+      expect(await countNodes('WithoutRetry')).toBe(1)
+    }, 60000)
+
+
+    it('should run transaction with retries on reactive failures', async () => {
+      if (protocolVersion < 4.0) {
+        return
+      }
+
+      const txcWork = new ConfigurableTransactionWork({
+        query: 'CREATE (:WithReactiveFailure) RETURN 7',
+        reactiveFailures: [
+          newError('service unavailable', SERVICE_UNAVAILABLE),
+          newError('session expired', SESSION_EXPIRED),
+          newError('transient error', 'Neo.TransientError.Transaction.NotStarted')
+        ]
+      })
+
+      const result = await session
+        .executeWrite(txc => txcWork.work(txc))
+        .pipe(materialize(), toArray())
+        .toPromise()
+      expect(result).toEqual([
+        Notification.createNext(7),
+        Notification.createComplete()
+      ])
+
+      expect(txcWork.invocations).toBe(4)
+      expect(await countNodes('WithReactiveFailure')).toBe(1)
+    }, 60000)
+
+    it('should run transaction with retries on synchronous failures', async () => {
+      if (protocolVersion < 4.0) {
+        return
+      }
+
+      const txcWork = new ConfigurableTransactionWork({
+        query: 'CREATE (:WithSyncFailure) RETURN 9',
+        syncFailures: [
+          newError('service unavailable', SERVICE_UNAVAILABLE),
+          newError('session expired', SESSION_EXPIRED),
+          newError('transient error', 'Neo.TransientError.Transaction.NotStarted')
+        ]
+      })
+
+      const result = await session
+        .executeWrite(txc => txcWork.work(txc))
+        .pipe(materialize(), toArray())
+        .toPromise()
+      expect(result).toEqual([
+        Notification.createNext(9),
+        Notification.createComplete()
+      ])
+
+      expect(txcWork.invocations).toBe(4)
+      expect(await countNodes('WithSyncFailure')).toBe(1)
+    }, 60000)
+
+    it('should fail on transactions that cannot be retried', async () => {
+      if (protocolVersion < 4.0) {
+        return
+      }
+
+      const txcWork = new ConfigurableTransactionWork({
+        query: 'UNWIND [10, 5, 0] AS x CREATE (:Hi) RETURN 10/x'
+      })
+
+      const result = await session
+        .executeWrite(txc => txcWork.work(txc))
+        .pipe(materialize(), toArray())
+        .toPromise()
+      expect(result).toEqual([
+        Notification.createNext(1),
+        Notification.createNext(2),
+        Notification.createError(jasmine.stringMatching(/\/ by zero/))
+      ])
+
+      expect(txcWork.invocations).toBe(1)
+      expect(await countNodes('Hi')).toBe(0)
+    }, 60000)
+
+    it('should fail even after a transient error', async () => {
+      if (protocolVersion < 4.0) {
+        return
+      }
+
+      const txcWork = new ConfigurableTransactionWork({
+        query: 'CREATE (:Person) RETURN 1',
+        syncFailures: [
+          newError(
+            'a transient error',
+            'Neo.TransientError.Transaction.NotStarted'
+          )
+        ],
+        reactiveFailures: [
+          newError('a database error', 'Neo.Database.Not.Started')
+        ]
+      })
+
+      const result = await session
+        .executeWrite(txc => txcWork.work(txc))
+        .pipe(materialize(), toArray())
+        .toPromise()
+      expect(result).toEqual([
+        Notification.createError(jasmine.stringMatching(/a database error/))
+      ])
+
+      expect(txcWork.invocations).toBe(2)
+      expect(await countNodes('Person')).toBe(0)
+    }, 60000)
+  })
+
   async function countNodes (label) {
     const session = driver.rxSession()
     return await session

packages/neo4j-driver/test/types/session-rx.test.ts

@@ -19,6 +19,7 @@
 
 import RxSession from '../../types/session-rx'
 import RxTransaction from '../../types/transaction-rx'
+import { RxManagedTransaction } from '../../types'
 import RxResult from '../../types/result-rx'
 import {
   Integer,
@@ -152,3 +153,27 @@ const observable6: Observable<number> = rxSession.writeTransaction(
   (tx: RxTransaction) => of(42),
   txConfig4
 )
+
+const observable7: Observable<number> = rxSession.executeRead(
+  (tx: RxManagedTransaction) => {
+    return of(10)
+  }
+)
+
+const observable8: Observable<string> = rxSession.executeRead(
+  (tx: RxManagedTransaction) => {
+    return of('42')
+  }
+)
+
+const observable9: Observable<number> = rxSession.executeWrite(
+  (tx: RxManagedTransaction) => {
+    return of(10)
+  }
+)
+
+const observable10: Observable<string> = rxSession.executeWrite(
+  (tx: RxManagedTransaction) => {
+    return of('42')
+  }
+)

packages/neo4j-driver/test/types/transaction-managed-rx.test.ts

@@ -0,0 +1,64 @@
+/**
+ * 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 RxManagedTransaction from '../../types/transaction-managed-rx'
+import { Record, ResultSummary } from 'neo4j-driver-core'
+import RxResult from '../../types/result-rx'
+import { Observable, of, Observer, throwError } from 'rxjs'
+import { concat, finalize, catchError } from 'rxjs/operators'
+
+const dummy: any = null
+
+const stringObserver: Observer<string> = {
+  next: value => console.log(value),
+  complete: () => console.log('complete'),
+  error: error => console.log(`error: ${error}`)
+}
+
+const keysObserver: Observer<string[]> = {
+  next: value => console.log(`keys: ${value}`),
+  complete: () => console.log('keys complete'),
+  error: error => console.log(`keys error: ${error}`)
+}
+
+const recordsObserver: Observer<Record> = {
+  next: value => console.log(`record: ${value}`),
+  complete: () => console.log('records complete'),
+  error: error => console.log(`records error: ${error}`)
+}
+
+const summaryObserver: Observer<ResultSummary> = {
+  next: value => console.log(`summary: ${value}`),
+  complete: () => console.log('summary complete'),
+  error: error => console.log(`summary error: ${error}`)
+}
+
+const tx: RxManagedTransaction = dummy
+
+const result1: RxResult = tx.run('RETURN 1')
+result1.keys().subscribe(keysObserver)
+result1.records().subscribe(recordsObserver)
+result1.consume().subscribe(summaryObserver)
+
+const result2: RxResult = tx.run('RETURN $value', { value: '42' })
+result2.keys().subscribe(keysObserver)
+result2.records().subscribe(recordsObserver)
+result2.consume().subscribe(summaryObserver)
+
+const isOpen: boolean = tx.isOpen()