moving transaction to core

Author: bigmontz

core/src/index.ts

@@ -66,6 +66,7 @@ import ResultSummary, {
 } from './result-summary'
 import Result, { QueryResult, ResultObserver } from './result'
 import ConnectionProvider from './connection-provider'
+import Transaction from './transaction'
 import * as internal from './internal' // todo: removed afterwards
 
 /**
@@ -123,7 +124,8 @@ const forExport = {
   Plan,
   ProfiledPlan,
   QueryStatistics,
-  Result
+  Result,
+  Transaction
 }
 
 export {
@@ -175,7 +177,8 @@ export {
   Result,
   QueryResult,
   ResultObserver,
-  ConnectionProvider
+  ConnectionProvider,
+  Transaction
 }
 
 export default forExport

core/src/internal/observers.ts

@@ -17,6 +17,7 @@
  * limitations under the License.
  */
 
+import { observer } from '.'
 import Record from '../record'
 import ResultSummary from '../result-summary'
 
@@ -34,7 +35,7 @@ interface StreamObserver {
    * to it's onError method, otherwise set instance variable _error.
    * @param {Object} error - An error object
    */
-  onError?: (error: Error) => void
+  onError: (error: Error) => void
   onCompleted?: (meta: any) => void
 }
 
@@ -58,9 +59,9 @@ interface ResultObserver {
 
   /**
    * Called when the result is fully received
-   * @param {ResultSummary} summary The result summary
+   * @param {ResultSummary| any} summary The result summary
    */
-  onCompleted?: (summary: ResultSummary) => void
+  onCompleted?: (summary: ResultSummary | any) => void
 
   /**
    * Called when some error occurs during the result proccess or query execution
@@ -105,4 +106,74 @@ export interface ResultStreamObserver extends StreamObserver {
   subscribe(observer: ResultObserver): void
 }
 
-// @todo implement observers
+export class CompletedObserver implements ResultStreamObserver {
+  subscribe(observer: ResultObserver): void {
+    apply(observer, observer.onKeys, [])
+    apply(observer, observer.onCompleted, {})
+  }
+
+  cancel(): void {
+    // do nothing
+  }
+
+  prepareToHandleSingleResponse(): void {
+    // do nothing
+  }
+
+  markCompleted(): void {
+    // do nothing
+  }
+
+  onError(error: Error): void {
+    // nothing to do, already finished
+    throw Error('CompletedObserver not supposed to call onError')
+  }
+}
+
+export class FailedObserver implements ResultStreamObserver {
+  private _error: Error
+  private _beforeError?: (error: Error) => void
+  private _observers: ResultObserver[]
+
+  constructor({
+    error,
+    onError
+  }: {
+    error: Error
+    onError?: (error: Error) => void | Promise<void>
+  }) {
+    this._error = error
+    this._beforeError = onError
+    this._observers = []
+    this.onError(error)
+  }
+
+  subscribe(observer: ResultObserver): void {
+    apply(observer, observer.onError, this._error)
+    this._observers.push(observer)
+  }
+
+  onError(error: Error): void {
+    Promise.resolve(apply(this, this._beforeError, error)).then(() =>
+      this._observers.forEach(o => apply(o, o.onError, error))
+    )
+  }
+
+  cancel(): void {
+    // do nothing
+  }
+
+  prepareToHandleSingleResponse(): void {
+    // do nothing
+  }
+
+  markCompleted(): void {
+    // do nothing
+  }
+}
+
+function apply<T>(thisArg: any, func?: (param: T) => void, param?: T): void {
+  if (func) {
+    func.bind(thisArg)(param)
+  }
+}

core/src/result.ts

@@ -21,6 +21,7 @@ import ResultSummary from './result-summary'
 import Record from './record'
 import { Query } from './types'
 import { observer, util, connectionHolder } from './internal'
+import { CompletedObserver, FailedObserver } from './internal/observers'
 
 const { EMPTY_CONNECTION_HOLDER } = connectionHolder
 
@@ -288,7 +289,9 @@ class Result implements Promise<QueryResult> {
     }
     observer.onError = onErrorWrapper
 
-    this._streamObserverPromise.then(o => o.subscribe(observer))
+    this._streamObserverPromise.then(o => {
+      return o.subscribe(observer)
+    })
   }
 
   /**

core/src/transacation.ts renamed to core/src/transaction.ts

@@ -93,11 +93,10 @@ class Transaction {
   _begin(bookmark: Bookmark | string | string[], txConfig: TxConfig) {
     this._connectionHolder
       .getConnection()
-      .then(conn => {
-        if (conn) {
-          this._onConnection()
-
-          return conn.protocol().beginTransaction({
+      .then(connection => {
+        this._onConnection()
+        if (connection) {
+          return connection.protocol().beginTransaction({
             bookmark: bookmark,
             txConfig: txConfig,
             mode: this._connectionHolder.mode(),
@@ -120,7 +119,7 @@ class Transaction {
    * @param {Object} parameters - Map with parameters to use in query
    * @return {Result} New Result
    */
-  run(query: Query, parameters: any): Result {
+  run(query: Query, parameters?: any): Result {
     const { validatedQuery, params } = validateQueryAndParameters(
       query,
       parameters
@@ -287,8 +286,8 @@ const _states = {
       const observerPromise = connectionHolder
         .getConnection()
         .then(conn => {
+          onConnection()
           if (conn) {
-            onConnection()
             return conn.protocol().run(query, parameters, {
               bookmark: Bookmark.empty(),
               txConfig: TxConfig.empty(),
@@ -507,7 +506,7 @@ function finishTransaction(
   onComplete: (metadata: any) => any,
   onConnection: () => any,
   pendingResults: Result[]
-) {
+): Result {
   const observerPromise = connectionHolder
     .getConnection()
     .then(connection => {
@@ -559,7 +558,7 @@ function newCompletedResult(
   query: Query,
   parameters: any,
   connectionHolder: ConnectionHolder = EMPTY_CONNECTION_HOLDER
-) {
+): Result {
   return new Result(
     Promise.resolve(observerPromise),
     query,

src/session.js

@@ -17,8 +17,7 @@
  * limitations under the License.
  */
 import { ResultStreamObserver, FailedObserver } from './internal/bolt'
-import Transaction from './transaction'
-import { newError, internal, Result } from 'neo4j-driver-core'
+import { newError, internal, Result, Transaction } from 'neo4j-driver-core'
 import ConnectionHolder from './internal/connection-holder'
 import { ACCESS_MODE_READ, ACCESS_MODE_WRITE } from './internal/constants'
 import TransactionExecutor from './internal/transaction-executor'

src/transaction-rx.js

@@ -18,7 +18,7 @@
  */
 import { Observable } from 'rxjs'
 import RxResult from './result-rx'
-import Transaction from './transaction'
+import Transaction from 'neo4j-driver-core'
 
 /**
  * A reactive transaction, which provides the same functionality as {@link Transaction} but through a Reactive API.