AsyncIterator API for Result consumption (#831)

The current driver implementation has only two options for record consumption, the default async/promise api and the reactive api. 

In the async side, there is no simple way to interact over the results before consuming all the data stream and back-pressure the consumption. Meanwhile in the reactive, the back-pressure is implemented  but this api usage depends on the knowledge of reactive programming and `RxJS`.  Another issue with our current reactive implementation is not having the way to configure the back-pressure or pause the stream consumption. 

The introduction of the Async Iterator is part of a group of improvements to make the driver more idiomatic and memory efficient.

Author: bigmontz

packages/bolt-connection/src/bolt/bolt-protocol-v1.js

@@ -278,7 +278,9 @@ export default class BoltProtocol {
       afterError,
       beforeComplete,
       afterComplete,
-      flush = true
+      flush = true,
+      highRecordWatermark = Number.MAX_VALUE,
+      lowRecordWatermark = Number.MAX_VALUE
     } = {}
   ) {
     const observer = new ResultStreamObserver({
@@ -288,7 +290,9 @@ export default class BoltProtocol {
       beforeError,
       afterError,
       beforeComplete,
-      afterComplete
+      afterComplete,
+      highRecordWatermark,
+      lowRecordWatermark
     })
 
     // bookmark and mode are ignored in this version of the protocol

packages/bolt-connection/src/bolt/bolt-protocol-v3.js

@@ -163,7 +163,9 @@ export default class BoltProtocol extends BoltProtocolV2 {
       afterError,
       beforeComplete,
       afterComplete,
-      flush = true
+      flush = true,
+      highRecordWatermark = Number.MAX_VALUE,
+      lowRecordWatermark = Number.MAX_VALUE
     } = {}
   ) {
     const observer = new ResultStreamObserver({
@@ -173,7 +175,9 @@ export default class BoltProtocol extends BoltProtocolV2 {
       beforeError,
       afterError,
       beforeComplete,
-      afterComplete
+      afterComplete,
+      highRecordWatermark,
+      lowRecordWatermark
     })
 
     // passing in a database name on this protocol version throws an error

packages/bolt-connection/src/bolt/bolt-protocol-v4x0.js

@@ -90,7 +90,9 @@ export default class BoltProtocol extends BoltProtocolV3 {
       afterComplete,
       flush = true,
       reactive = false,
-      fetchSize = FETCH_ALL
+      fetchSize = FETCH_ALL,
+      highRecordWatermark = Number.MAX_VALUE,
+      lowRecordWatermark = Number.MAX_VALUE
     } = {}
   ) {
     const observer = new ResultStreamObserver({
@@ -104,7 +106,9 @@ export default class BoltProtocol extends BoltProtocolV3 {
       beforeError,
       afterError,
       beforeComplete,
-      afterComplete
+      afterComplete,
+      highRecordWatermark,
+      lowRecordWatermark
     })
 
     // passing impersonated user on this protocol version throws an error

packages/bolt-connection/src/bolt/bolt-protocol-v4x4.js

@@ -84,7 +84,9 @@ export default class BoltProtocol extends BoltProtocolV43 {
       afterComplete,
       flush = true,
       reactive = false,
-      fetchSize = FETCH_ALL
+      fetchSize = FETCH_ALL,
+      highRecordWatermark = Number.MAX_VALUE,
+      lowRecordWatermark = Number.MAX_VALUE
     } = {}
   ) {
     const observer = new ResultStreamObserver({
@@ -98,7 +100,9 @@ export default class BoltProtocol extends BoltProtocolV43 {
       beforeError,
       afterError,
       beforeComplete,
-      afterComplete
+      afterComplete,
+      highRecordWatermark,
+      lowRecordWatermark
     })
 
     const flushRun = reactive

packages/bolt-connection/src/bolt/stream-observers.js

@@ -68,7 +68,9 @@ class ResultStreamObserver extends StreamObserver {
     afterKeys,
     beforeComplete,
     afterComplete,
-    server
+    server,
+    highRecordWatermark = Number.MAX_VALUE,
+    lowRecordWatermark = Number.MAX_VALUE
   } = {}) {
     super()
 
@@ -94,8 +96,32 @@ class ResultStreamObserver extends StreamObserver {
     this._discardFunction = discardFunction
     this._discard = false
     this._fetchSize = fetchSize
+    this._lowRecordWatermark = lowRecordWatermark
+    this._highRecordWatermark = highRecordWatermark
     this._setState(reactive ? _states.READY : _states.READY_STREAMING)
-    this._setupAuoPull(fetchSize)
+    this._setupAutoPull()
+    this._paused = false;
+  }
+
+  /**
+   * Pause the record consuming
+   *
+   * This function will supend the record consuming. It will not cancel the stream and the already
+   * requested records will be sent to the subscriber.
+   */
+  pause () {
+    this._paused = true
+  }
+
+  /**
+   * Resume the record consuming
+   *
+   * This function will resume the record consuming fetching more records from the server.
+   */
+  resume () {
+    this._paused = false
+    this._setupAutoPull(true)
+    this._state.pull(this)
   }
 
   /**
@@ -342,16 +368,21 @@ class ResultStreamObserver extends StreamObserver {
 
   _handleStreaming () {
     if (this._head && this._observers.some(o => o.onNext || o.onCompleted)) {
-      if (this._discard) {
-        this._discardFunction(this._queryId, this)
-        this._setState(_states.STREAMING)
-      } else if (this._autoPull) {
-        this._moreFunction(this._queryId, this._fetchSize, this)
-        this._setState(_states.STREAMING)
+      if (!this._paused && (this._discard || this._autoPull)) {
+        this._more()
       }
     }
   }
 
+  _more () {
+    if (this._discard) {
+      this._discardFunction(this._queryId, this)
+    } else {
+      this._moreFunction(this._queryId, this._fetchSize, this)
+    }
+    this._setState(_states.STREAMING)
+  }
+
   _storeMetadataForCompletion (meta) {
     const keys = Object.keys(meta)
     let index = keys.length
@@ -367,15 +398,8 @@ class ResultStreamObserver extends StreamObserver {
     this._state = state
   }
 
-  _setupAuoPull (fetchSize) {
+  _setupAutoPull () {
     this._autoPull = true
-    if (fetchSize === FETCH_ALL) {
-      this._lowRecordWatermark = Number.MAX_VALUE // we shall always lower than this number to enable auto pull
-      this._highRecordWatermark = Number.MAX_VALUE // we shall never reach this number to disable auto pull
-    } else {
-      this._lowRecordWatermark = 0.3 * fetchSize
-      this._highRecordWatermark = 0.7 * fetchSize
-    }
   }
 }
 
@@ -575,7 +599,8 @@ const _states = {
     },
     name: () => {
       return 'READY_STREAMING'
-    }
+    },
+    pull: () => {}
   },
   READY: {
     // reactive start state
@@ -590,7 +615,8 @@ const _states = {
     },
     name: () => {
       return 'READY'
-    }
+    },
+    pull: streamObserver => streamObserver._more()
   },
   STREAMING: {
     onSuccess: (streamObserver, meta) => {
@@ -605,20 +631,23 @@ const _states = {
     },
     name: () => {
       return 'STREAMING'
-    }
+    },
+    pull: () => {}
   },
   FAILED: {
     onError: error => {
       // more errors are ignored
     },
     name: () => {
       return 'FAILED'
-    }
+    },
+    pull: () => {}
   },
   SUCCEEDED: {
     name: () => {
       return 'SUCCEEDED'
-    }
+    },
+    pull: () => {}
   }
 }
 

packages/bolt-connection/test/bolt/bolt-protocol-v1.test.js

@@ -338,4 +338,26 @@ describe('#unit BoltProtocolV1', () => {
       }
     )
   })
+
+  describe('watermarks', () => {
+    it('.run() should configure watermarks', () => {
+      const recorder = new utils.MessageRecordingConnection()
+      const protocol = utils.spyProtocolWrite(
+        new BoltProtocolV1(recorder, null, false)
+      )
+
+      const query = 'RETURN $x, $y'
+      const parameters = { x: 'x', y: 'y' }
+      const observer = protocol.run(query, parameters, {
+        bookmark: Bookmark.empty(),
+        txConfig: TxConfig.empty(),
+        mode: WRITE,
+        lowRecordWatermark: 100,
+        highRecordWatermark: 200,
+      })
+
+      expect(observer._lowRecordWatermark).toEqual(100)
+      expect(observer._highRecordWatermark).toEqual(200)
+    })
+  })
 })

packages/bolt-connection/test/bolt/bolt-protocol-v2.test.js

@@ -92,4 +92,23 @@ describe('#unit BoltProtocolV2', () => {
       })
     })
   })
+
+  describe('watermarks', () => {
+    it('.run() should configure watermarks', () => {
+      const recorder = new utils.MessageRecordingConnection()
+      const protocol = utils.spyProtocolWrite(
+        new BoltProtocolV2(recorder, null, false)
+      )
+
+      const query = 'RETURN $x, $y'
+      const parameters = { x: 'x', y: 'y' }
+      const observer = protocol.run(query, parameters, {
+        lowRecordWatermark: 100,
+        highRecordWatermark: 200,
+      })
+
+      expect(observer._lowRecordWatermark).toEqual(100)
+      expect(observer._highRecordWatermark).toEqual(200)
+    })
+  })
 })

packages/bolt-connection/test/bolt/bolt-protocol-v3.test.js

@@ -295,6 +295,27 @@ describe('#unit BoltProtocolV3', () => {
       }
     )
   })
+
+  describe('watermarks', () => {
+    it('.run() should configure watermarks', () => {
+      const recorder = new utils.MessageRecordingConnection()
+      const protocol = utils.spyProtocolWrite(
+        new BoltProtocolV3(recorder, null, false)
+      )
+
+      const query = 'RETURN $x, $y'
+      const parameters = { x: 'x', y: 'y' }
+      const observer = protocol.run(query, parameters, {
+        bookmark: Bookmark.empty(),
+        txConfig: TxConfig.empty(),
+        lowRecordWatermark: 100,
+        highRecordWatermark: 200,
+      })
+
+      expect(observer._lowRecordWatermark).toEqual(100)
+      expect(observer._highRecordWatermark).toEqual(200)
+    })
+  })
 })
 
 class SpiedBoltProtocolV3 extends BoltProtocolV3 {

packages/bolt-connection/test/bolt/bolt-protocol-v4x0.test.js

@@ -214,6 +214,27 @@ describe('#unit BoltProtocolV4x0', () => {
       }
     )
   })
+
+  describe('watermarks', () => {
+    it('.run() should configure watermarks', () => {
+      const recorder = new utils.MessageRecordingConnection()
+      const protocol = utils.spyProtocolWrite(
+        new BoltProtocolV4x0(recorder, null, false)
+      )
+
+      const query = 'RETURN $x, $y'
+      const parameters = { x: 'x', y: 'y' }
+      const observer = protocol.run(query, parameters, {
+        bookmark: Bookmark.empty(),
+        txConfig: TxConfig.empty(),
+        lowRecordWatermark: 100,
+        highRecordWatermark: 200,
+      })
+
+      expect(observer._lowRecordWatermark).toEqual(100)
+      expect(observer._highRecordWatermark).toEqual(200)
+    })
+  })
 })
 
 class SpiedBoltProtocolV4x0 extends BoltProtocolV4x0 {

packages/bolt-connection/test/bolt/bolt-protocol-v4x1.test.js

@@ -19,6 +19,12 @@
 
 import BoltProtocolV4x1 from '../../src/bolt/bolt-protocol-v4x1'
 import utils from '../test-utils'
+import { internal } from 'neo4j-driver-core'
+
+const {
+  txConfig: { TxConfig },
+  bookmark: { Bookmark }
+} = internal
 
 describe('#unit BoltProtocolV4x1', () => {
   describe('Bolt v4.4', () => {
@@ -82,4 +88,25 @@ describe('#unit BoltProtocolV4x1', () => {
       }
     )
   })
+
+  describe('watermarks', () => {
+    it('.run() should configure watermarks', () => {
+      const recorder = new utils.MessageRecordingConnection()
+      const protocol = utils.spyProtocolWrite(
+        new BoltProtocolV4x1(recorder, null, false)
+      )
+
+      const query = 'RETURN $x, $y'
+      const parameters = { x: 'x', y: 'y' }
+      const observer = protocol.run(query, parameters, {
+        bookmark: Bookmark.empty(),
+        txConfig: TxConfig.empty(),
+        lowRecordWatermark: 100,
+        highRecordWatermark: 200,
+      })
+
+      expect(observer._lowRecordWatermark).toEqual(100)
+      expect(observer._highRecordWatermark).toEqual(200)
+    })
+  })
 })