RxJS 7

README.md

@@ -222,7 +222,7 @@ rxSession
   .records()
   .pipe(
     map(record => record.get('name')),
-    concat(rxSession.close())
+    concatWith(rxSession.close())
   )
   .subscribe({
     next: data => console.log(data),
@@ -373,7 +373,7 @@ try {
 rxSession
   .beginTransaction()
   .pipe(
-    flatMap(txc =>
+    mergeMap(txc =>
       concat(
         txc
           .run(

package.json

@@ -108,6 +108,6 @@
     "@babel/runtime": "^7.5.5",
     "neo4j-driver-bolt-connection": "file:./bolt-connection",
     "neo4j-driver-core": "file:./core",
-    "rxjs": "^6.6.3"
+    "rxjs": "^7.3.0"
   }
 }

src/internal/retry-logic-rx.js

@@ -19,7 +19,7 @@
 
 import { newError, error, internal } from 'neo4j-driver-core'
 import { Observable, throwError, of } from 'rxjs'
-import { retryWhen, flatMap, delay } from 'rxjs/operators'
+import { retryWhen, mergeMap as flatMap, delay } from 'rxjs/operators'
 
 const {
   logger: {
@@ -82,7 +82,7 @@ export default class RxRetryLogic {
         return failedWork.pipe(
           flatMap(err => {
             if (!canRetryOn(err)) {
-              return throwError(err)
+              return throwError(() => err)
             }
 
             handledExceptions.push(err)
@@ -98,7 +98,7 @@ export default class RxRetryLogic {
 
               error.seenErrors = handledExceptions
 
-              return throwError(error)
+              return throwError(() => error)
             }
 
             const nextDelayDuration = this._computeNextDelay(delayDuration)

src/result-rx.js

@@ -18,7 +18,7 @@
  */
 import { newError, Record, ResultSummary } from 'neo4j-driver-core'
 import { Observable, Subject, ReplaySubject, from } from 'rxjs'
-import { flatMap, publishReplay, refCount, shareReplay } from 'rxjs/operators'
+import { mergeMap as flatMap, shareReplay } from 'rxjs/operators'
 
 const States = {
   READY: 0,
@@ -36,13 +36,12 @@ export default class RxResult {
    * @param {Observable<Result>} result - An observable of single Result instance to relay requests.
    */
   constructor (result) {
-    const replayedResult = result.pipe(publishReplay(1), refCount())
+    const replayedResult = result.pipe(shareReplay(1))
 
     this._result = replayedResult
     this._keys = replayedResult.pipe(
       flatMap(r => from(r.keys())),
-      publishReplay(1),
-      refCount()
+      shareReplay(1)
     )
     this._records = new Subject()
     this._summary = new ReplaySubject()

src/session-rx.js

@@ -16,8 +16,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { defer, Observable, throwError } from 'rxjs'
-import { flatMap, catchError, concat } from 'rxjs/operators'
+import { defer, from, Observable, throwError } from 'rxjs'
+import {
+  mergeMap as flatMap,
+  catchError,
+  concatWith as concat
+} from 'rxjs/operators'
 import RxResult from './result-rx'
 import { Session, internal } from 'neo4j-driver-core'
 import RxTransaction from './transaction-rx'
@@ -56,24 +60,15 @@ export default class RxSession {
    */
   run (query, parameters, transactionConfig) {
     return new RxResult(
-      new Observable(observer => {
-        try {
-          observer.next(this._session.run(query, parameters, transactionConfig))
-          observer.complete()
-        } catch (err) {
-          observer.error(err)
-        }
-
-        return () => {}
-      })
+      defer(() => [this._session.run(query, parameters, transactionConfig)])
     )
   }
 
   /**
    * Starts a new explicit transaction with the provided transaction configuration.
    *
    * @public
-   * @param {TransactionConfig} transactionConfig - Configuration for the new transaction.
+   * @param {TransactionConfig} [transactionConfig] - Configuration for the new transaction.
    * @returns {Observable<RxTransaction>} - A reactive stream that will generate at most **one** RxTransaction instance.
    */
   beginTransaction (transactionConfig) {
@@ -85,7 +80,7 @@ export default class RxSession {
    * transaction configuration.
    * @public
    * @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.
+   * @param {TransactionConfig} [transactionConfig] - Configuration for the enclosing transaction created by the driver.
    * @returns {Observable} - A reactive stream returned by the unit of work.
    */
   readTransaction (work, transactionConfig) {
@@ -111,14 +106,7 @@ export default class RxSession {
    * @returns {Observable} - An empty reactive stream
    */
   close () {
-    return new Observable(observer => {
-      this._session
-        .close()
-        .then(() => {
-          observer.complete()
-        })
-        .catch(err => observer.error(err))
-    })
+    return from(this._session.close())
   }
 
   /**
@@ -140,47 +128,32 @@ export default class RxSession {
    * @private
    */
   _beginTransaction (accessMode, transactionConfig) {
-    let txConfig = TxConfig.empty()
-    if (transactionConfig) {
-      txConfig = new TxConfig(transactionConfig)
-    }
-
-    return new Observable(observer => {
-      try {
-        observer.next(
-          new RxTransaction(
-            this._session._beginTransaction(accessMode, txConfig)
-          )
-        )
-        observer.complete()
-      } catch (err) {
-        observer.error(err)
-      }
+    const txConfig = transactionConfig
+      ? new TxConfig(transactionConfig)
+      : TxConfig.empty()
 
-      return () => {}
-    })
+    return defer(() => [
+      new RxTransaction(this._session._beginTransaction(accessMode, txConfig))
+    ])
   }
 
   /**
    * @private
    */
   _runTransaction (accessMode, work, transactionConfig) {
-    let txConfig = TxConfig.empty()
-    if (transactionConfig) {
-      txConfig = new TxConfig(transactionConfig)
-    }
-
     return this._retryLogic.retry(
       this._beginTransaction(accessMode, transactionConfig).pipe(
         flatMap(txc =>
           defer(() => {
             try {
               return work(txc)
             } catch (err) {
-              return throwError(err)
+              return throwError(() => err)
             }
           }).pipe(
-            catchError(err => txc.rollback().pipe(concat(throwError(err)))),
+            catchError(err =>
+              txc.rollback().pipe(concat(throwError(() => err)))
+            ),
             concat(txc.commit())
           )
         )

src/transaction-rx.js

@@ -16,7 +16,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { Observable } from 'rxjs'
+import { defer, from, Observable } from 'rxjs'
 import RxResult from './result-rx'
 import Transaction from 'neo4j-driver-core'
 
@@ -41,53 +41,27 @@ 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 => {
-        try {
-          observer.next(this._txc.run(query, parameters))
-          observer.complete()
-        } catch (err) {
-          observer.error(err)
-        }
-
-        return () => {}
-      })
-    )
+    return new RxResult(defer(() => [this._txc.run(query, parameters)]))
   }
 
   /**
    *  Commits the transaction.
    *
    * @public
-   * @returns {Observable} - An empty observable
+   * @returns {Observable<void>} - An empty observable
    */
   commit () {
-    return new Observable(observer => {
-      this._txc
-        .commit()
-        .then(() => {
-          observer.complete()
-        })
-        .catch(err => observer.error(err))
-    })
+    return from(this._txc.commit())
   }
 
   /**
    *  Rolls back the transaction.
    *
    * @public
-   * @returns {Observable} - An empty observable
+   * @returns {Observable<void>} - An empty observable
    */
   rollback () {
-    return new Observable(observer => {
-      this._txc
-        .rollback()
-        .then(() => {
-          observer.complete()
-        })
-        .catch(err => observer.error(err))
-    })
+    return from(this._txc.rollback())
   }
 }

test/examples.test.js

@@ -21,7 +21,7 @@ import neo4j from '../src'
 import sharedNeo4j from './internal/shared-neo4j'
 import { ServerVersion, VERSION_4_0_0 } from '../src/internal/server-version'
 import { map, materialize, toArray } from 'rxjs/operators'
-import { Notification } from 'rxjs'
+import { lastValueFrom } from 'rxjs'
 
 /**
  * The tests below are examples that get pulled into the Driver Manual using the tags inside the tests.
@@ -143,12 +143,9 @@ describe('#integration examples', () => {
     }
 
     // end::rx-autocommit-transaction[]
-    const result = await readProductTitles().toPromise()
+    const result = await lastValueFrom(readProductTitles())
 
-    expect(result).toEqual([
-      Notification.createNext('Product-0'),
-      Notification.createComplete()
-    ])
+    expect(result).toEqual([{ kind: 'N', value: 'Product-0' }, { kind: 'C' }])
   }, 60000)
 
   it('basic auth example', async () => {
@@ -623,11 +620,11 @@ describe('#integration examples', () => {
       )
     // end::rx-result-consume[]
 
-    const people = await result.toPromise()
+    const people = await lastValueFrom(result)
     expect(people).toEqual([
-      Notification.createNext('Alice'),
-      Notification.createNext('Bob'),
-      Notification.createComplete()
+      { kind: 'N', value: 'Alice' },
+      { kind: 'N', value: 'Bob' },
+      { kind: 'C' }
     ])
   }, 60000)
 
@@ -802,11 +799,11 @@ describe('#integration examples', () => {
     )
     // end::rx-transaction-function[]
 
-    const people = await result.toPromise()
+    const people = await lastValueFrom(result)
     expect(people).toEqual([
-      Notification.createNext('Infinity Gauntlet'),
-      Notification.createNext('Mjölnir'),
-      Notification.createComplete()
+      { kind: 'N', value: 'Infinity Gauntlet' },
+      { kind: 'N', value: 'Mjölnir' },
+      { kind: 'C' }
     ])
   }, 60000)
 

test/internal/retry-logic-rx.test.js

@@ -245,7 +245,7 @@ describe('#unit-rx retrylogic', () => {
         clock.tick(delayBy)
       }
       if (index < errors.length) {
-        return throwError(errors[index++])
+        return throwError(() => errors[index++])
       } else {
         return of(value)
       }