AsynIterator API for Result consumption

Author: bigmontz

packages/core/src/result.ts

@@ -192,6 +192,46 @@ class Result implements Promise<QueryResult> {
     return this._p
   }
 
+  async* ayncInterator(): any {
+    function createResolvablePromise (): any {
+      const resolvablePromise: any = {}
+      resolvablePromise.promise = new Promise((resolve, reject) => {
+        resolvablePromise.resolve = resolve
+        resolvablePromise.reject = reject
+      });
+      return resolvablePromise;
+    }
+
+    const observer = {
+      _buffer: [createResolvablePromise()],
+      onNext: (record: Record) => {
+        observer._buffer[observer._buffer.length - 1].resolve({ record, done: false });
+        observer._buffer.push(createResolvablePromise());
+      },
+      onCompleted: (summary: ResultSummary) => {
+        observer._buffer[observer._buffer.length - 1].resolve({ summary, done: true });
+      },
+      onError: (error: Error) => {
+        observer._buffer[observer._buffer.length - 1].reject(error);
+      },
+      consume: async () => {
+        const value = await observer._buffer[0].promise
+        observer._buffer.shift();
+        return value
+      }
+    }
+
+    this.subscribe(observer)
+
+    while(true) {
+      const value = await observer.consume()
+      if (value.done) {
+        return value.summary;
+      } 
+      yield value.record
+    }
+  }
+
   /**
    * Waits for all results and calls the passed in function with the results.
    *

packages/testkit-backend/src/context.js

@@ -7,6 +7,7 @@ export default class Context {
     this._resolverRequests = {}
     this._resultObservers = {}
     this._errors = {}
+    this._results = {}
   }
 
   addDriver (driver) {
@@ -44,6 +45,18 @@ export default class Context {
     return id
   }
 
+  addResult (result) {
+    return this._add(this._results, result)
+  }
+
+  removeResult (id) {
+    delete this._results[id]
+  }
+
+  getResult (id) {
+    return this._results[id]
+  }
+
   getDriver (id) {
     return this._drivers[id]
   }

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

@@ -2,6 +2,7 @@ import neo4j from 'neo4j'
 import ResultObserver from './result-observer.js'
 import { cypherToNative, nativeToCypher } from './cypher-native-binders.js'
 import { shouldRunTest } from './skipped-tests'
+const USE_ASYNC = true
 
 export function NewDriver (context, data, { writeResponse }) {
   const {
@@ -89,49 +90,89 @@ export function SessionRun (context, data, wire) {
     .catch(_ => null)
     .then(_ => {
       const result = session.run(cypher, params, { metadata, timeout })
-      const resultObserver = new ResultObserver({ sessionId, result })
-      result.subscribe(resultObserver)
-      const id = context.addResultObserver(resultObserver)
+      let id
+      if (USE_ASYNC) {
+        id = context.addResult(result)
+      } else {
+        const resultObserver = new ResultObserver({ sessionId, result })
+        result.subscribe(resultObserver)
+        id = context.addResultObserver(resultObserver)
+      }
       wire.writeResponse('Result', { id })
     })
 }
 
 export function ResultNext (context, data, wire) {
   const { resultId } = data
-  const resultObserver = context.getResultObserver(resultId)
-  const nextPromise = resultObserver.next()
-  nextPromise
-    .then(rec => {
-      if (rec) {
-        const values = Array.from(rec.values()).map(nativeToCypher)
+  if (USE_ASYNC) {
+    const result = context.getResult(resultId)
+    if (!("recordIt" in result)) {
+      result.recordIt = result.ayncInterator()
+    }
+    result.recordIt.next().then(({ value, done }) => {
+      if (done) {
+        wire.writeResponse('NullRecord', null)
+      } else {
+        const values = Array.from(value.values()).map(nativeToCypher)
         wire.writeResponse('Record', {
           values: values
         })
-      } else {
-        wire.writeResponse('NullRecord', null)
       }
-    })
-    .catch(e => {
+    }).catch(e => {
       console.log('got some err: ' + JSON.stringify(e))
       wire.writeError(e)
-    })
+    });
+  } else {
+    const resultObserver = context.getResultObserver(resultId)
+    const nextPromise = resultObserver.next()
+    nextPromise
+      .then(rec => {
+        if (rec) {
+          const values = Array.from(rec.values()).map(nativeToCypher)
+          wire.writeResponse('Record', {
+            values: values
+          })
+        } else {
+          wire.writeResponse('NullRecord', null)
+        }
+      })
+      .catch(e => {
+        console.log('got some err: ' + JSON.stringify(e))
+        wire.writeError(e)
+      })
+  }
+  
 }
 
 export function ResultConsume (context, data, wire) {
   const { resultId } = data
-  const resultObserver = context.getResultObserver(resultId)
-  resultObserver
-    .completitionPromise()
-    .then(summary => {
+  if (USE_ASYNC) {
+    const result = context.getResult(resultId)
+    result.summary().then(summary => {
+      console.log(summary);
       wire.writeResponse('Summary', {
         ...summary,
         serverInfo: {
           agent: summary.server.agent,
-          protocolVersion: summary.server.protocolVersion.toFixed(1)
+          protocolVersion: summary.server.protocolVersion? summary.server.protocolVersion.toFixed(1) : 0
         }
       })
-    })
-    .catch(e => wire.writeError(e))
+    }).catch(e => wire.writeError(e))
+  } else {
+    const resultObserver = context.getResultObserver(resultId)
+    resultObserver
+      .completitionPromise()
+      .then(summary => {
+        wire.writeResponse('Summary', {
+          ...summary,
+          serverInfo: {
+            agent: summary.server.agent,
+            protocolVersion: summary.server.protocolVersion.toFixed(1)
+          }
+        })
+      })
+      .catch(e => wire.writeError(e))
+  }
 }
 
 export function SessionReadTransaction (context, data, wire) {
@@ -158,9 +199,15 @@ export function TransactionRun (context, data, wire) {
     }
   }
   const result = tx.tx.run(cypher, params)
-  const resultObserver = new ResultObserver({ result })
-  result.subscribe(resultObserver)
-  const id = context.addResultObserver(resultObserver)
+
+  let id
+  if (USE_ASYNC) {
+    id = context.addResult(result)
+  } else {
+    const resultObserver = new ResultObserver({ result })
+    result.subscribe(resultObserver)
+    id = context.addResultObserver(resultObserver)
+  }
   wire.writeResponse('Result', { id })
 }