Merge pull request #631 from bigmontz/testkit_wait_for_result_complete

testkit: waiting for the current result get completed

Author: bigmontz

testkit-backend/main.js

@@ -124,14 +124,22 @@ class Backend {
               params[key] = cypherToNative(value)
             }
           }
-          const result = session.run(cypher, params)
-          this._id++
-          const resultObserver = new ResultObserver()
-          result.subscribe(resultObserver)
-          this._resultObservers[this._id] = resultObserver
-          this._writeResponse('Result', {
-            id: this._id
-          })
+
+          const observers = Object.values(this._resultObservers).filter(
+            obs => obs.sessionId === sessionId
+          )
+          Promise.all(observers.map(obs => obs.completitionPromise()))
+            .catch(_ => null)
+            .then(_ => {
+              this._id++
+              const result = session.run(cypher, params)
+              const resultObserver = new ResultObserver({ sessionId })
+              result.subscribe(resultObserver)
+              this._resultObservers[this._id] = resultObserver
+              this._writeResponse('Result', {
+                id: this._id
+              })
+            })
         }
         break
 
@@ -191,7 +199,7 @@ class Backend {
           }
           const result = tx.tx.run(cypher, params)
           this._id++
-          const resultObserver = new ResultObserver()
+          const resultObserver = new ResultObserver({})
           result.subscribe(resultObserver)
           this._resultObservers[this._id] = resultObserver
           this._writeResponse('Result', {

testkit-backend/result-observer.js

@@ -1,7 +1,8 @@
 const neo4j = require('neo4j-driver')
 
 export class ResultObserver {
-  constructor () {
+  constructor ({ sessionId }) {
+    this.sessionId = sessionId
     this.keys = null
     this._stream = []
     this.summary = null
@@ -11,6 +12,7 @@ export class ResultObserver {
     this.onNext = this.onNext.bind(this)
     this.onCompleted = this.onCompleted.bind(this)
     this.onError = this.onError.bind(this)
+    this._completitionPromise = null
   }
 
   onKeys (keys) {
@@ -25,11 +27,15 @@ export class ResultObserver {
   onCompleted (summary) {
     this._summary = summary
     this._fulfill()
+    this._resolve(this._completitionPromise, summary)
+    this._completitionPromise = null
   }
 
   onError (e) {
     this._stream.push(e)
     this._fulfill()
+    this._reject(this._completitionPromise, e)
+    this._completitionPromise = null
   }
 
   // Returns a promise, only one outstanding next!
@@ -43,6 +49,21 @@ export class ResultObserver {
     })
   }
 
+  completitionPromise () {
+    return new Promise((resolve, reject) => {
+      if (this._summary) {
+        resolve(this._summary)
+      } else if (this._err) {
+        reject(this._err)
+      } else {
+        this._completitionPromise = {
+          resolve,
+          reject
+        }
+      }
+    })
+  }
+
   _fulfill () {
     if (!this._promise) {
       return
@@ -76,4 +97,16 @@ export class ResultObserver {
       this._promise = null
     }
   }
+
+  _resolve (promise, data) {
+    if (promise) {
+      promise.resolve(data)
+    }
+  }
+
+  _reject (promise, err) {
+    if (promise) {
+      promise.reject(err)
+    }
+  }
 }