Initial reactive session and transaction

Author: ali-ince

src/driver.js

@@ -17,20 +17,19 @@
  * limitations under the License.
  */
 
-import Session from './session'
-import Pool from './internal/pool'
-import ChannelConnection from './internal/connection-channel'
-import { newError, SERVICE_UNAVAILABLE } from './error'
-import DirectConnectionProvider from './internal/connection-provider-direct'
+import { newError } from './error'
+import ConnectionProvider from './internal/connection-provider'
 import Bookmark from './internal/bookmark'
+import DirectConnectionProvider from './internal/connection-provider-direct'
 import ConnectivityVerifier from './internal/connectivity-verifier'
-import PoolConfig, {
+import { ACCESS_MODE_READ, ACCESS_MODE_WRITE } from './internal/constants'
+import Logger from './internal/logger'
+import {
   DEFAULT_ACQUISITION_TIMEOUT,
   DEFAULT_MAX_SIZE
 } from './internal/pool-config'
-import Logger from './internal/logger'
-import ConnectionErrorHandler from './internal/connection-error-handler'
-import { ACCESS_MODE_READ, ACCESS_MODE_WRITE } from './internal/constants'
+import Session from './session'
+import RxSession from './session-rx'
 
 const DEFAULT_MAX_CONNECTION_LIFETIME = 60 * 60 * 1000 // 1 hour
 
@@ -121,17 +120,38 @@ class Driver {
    * it is closed, the underlying connection will be released to the connection
    * pool and made available for others to use.
    *
-   * @param {string} [defaultAccessMode=WRITE] the access mode of this session, allowed values are {@link READ} and {@link WRITE}.
-   * @param {string|string[]} [bookmarks=null] the initial reference or references to some previous
+   * @param {Object} args -
+   * @param {string} args.defaultAccessMode='WRITE' - the access mode of this session, allowed values are {@link READ} and {@link WRITE}.
+   * @param {string|string[]} args.bookmarks - the initial reference or references to some previous
    * transactions. Value is optional and absence indicates that that the bookmarks do not exist or are unknown.
-   * @param {string} [database=''] the database this session will connect to.
+   * @param {string} args.database='' - the database this session will connect to.
    * @return {Session} new session.
    */
   session ({
     defaultAccessMode = WRITE,
     bookmarks: bookmarkOrBookmarks,
     database = ''
   } = {}) {
+    return this._newSession({
+      defaultAccessMode,
+      bookmarkOrBookmarks,
+      database,
+      reactive: false
+    })
+  }
+
+  rxSession ({ defaultAccessMode = WRITE, bookmarks, database = '' } = {}) {
+    return new RxSession(
+      this._newSession({
+        defaultAccessMode,
+        bookmarks,
+        database,
+        reactive: true
+      })
+    )
+  }
+
+  _newSession ({ defaultAccessMode, bookmarkOrBookmarks, database, reactive }) {
     const sessionMode = Driver._validateSessionMode(defaultAccessMode)
     const connectionProvider = this._getOrCreateConnectionProvider()
     const bookmark = bookmarkOrBookmarks
@@ -142,7 +162,8 @@ class Driver {
       database,
       connectionProvider,
       bookmark,
-      config: this._config
+      config: this._config,
+      reactive
     })
   }
 

src/internal/bolt-protocol-v4.js

@@ -63,11 +63,15 @@ export default class BoltProtocol extends BoltProtocolV3 {
       afterError,
       beforeComplete,
       afterComplete,
-      flush = true
+      flush = true,
+      reactive = false
     } = {}
   ) {
     const observer = new ResultStreamObserver({
       connection: this._connection,
+      reactive: reactive,
+      moreFunction: reactive ? this._requestMore : this._noOp,
+      discardFunction: reactive ? this._requestDiscard : this._noOp,
       beforeKeys,
       afterKeys,
       beforeError,
@@ -76,6 +80,7 @@ export default class BoltProtocol extends BoltProtocolV3 {
       afterComplete
     })
 
+    const flushRun = reactive
     this._connection.write(
       RequestMessage.runWithMetadata(statement, parameters, {
         bookmark,
@@ -84,10 +89,23 @@ export default class BoltProtocol extends BoltProtocolV3 {
         mode
       }),
       observer,
-      false
+      flushRun && flush
     )
-    this._connection.write(RequestMessage.pull(), observer, flush)
+
+    if (!reactive) {
+      this._connection.write(RequestMessage.pull(), observer, flush)
+    }
 
     return observer
   }
+
+  _requestMore (connection, stmtId, n, observer) {
+    connection.write(RequestMessage.pull({ stmtId, n }), observer, true)
+  }
+
+  _requestDiscard (connection, stmtId, observer) {
+    connection.write(RequestMessage.discard({ stmtId }), observer, true)
+  }
+
+  _noOp () {}
 }

src/internal/request-message.js

@@ -185,7 +185,7 @@ export default class RequestMessage {
    * @return {RequestMessage} the PULL message.
    */
   static pull ({ stmtId = NO_STATEMENT_ID, n = ALL } = {}) {
-    const metadata = buildStreamMetadata(stmtId, n)
+    const metadata = buildStreamMetadata(stmtId || NO_STATEMENT_ID, n || ALL)
     return new RequestMessage(
       PULL,
       [metadata],
@@ -200,7 +200,7 @@ export default class RequestMessage {
    * @return {RequestMessage} the PULL message.
    */
   static discard ({ stmtId = NO_STATEMENT_ID, n = ALL } = {}) {
-    const metadata = buildStreamMetadata(stmtId, n)
+    const metadata = buildStreamMetadata(stmtId || NO_STATEMENT_ID, n || ALL)
     return new RequestMessage(
       DISCARD,
       [metadata],
@@ -246,7 +246,7 @@ function buildTxMetadata (bookmark, txConfig, database, mode) {
 function buildStreamMetadata (stmtId, n) {
   const metadata = { n: int(n) }
   if (stmtId !== NO_STATEMENT_ID) {
-    metadata['stmt_id'] = int(stmtId)
+    metadata['qid'] = int(stmtId)
   }
   return metadata
 }

src/internal/stream-observers.js

@@ -20,8 +20,9 @@ import Record from '../record'
 import Connection from './connection'
 import { newError, PROTOCOL_ERROR } from '../error'
 import { isString } from './util'
+import Integer from '../integer'
 
-const DefaultBatchSize = 100
+const DefaultBatchSize = 50
 
 class StreamObserver {
   onNext (rawRecord) {}
@@ -45,10 +46,11 @@ class ResultStreamObserver extends StreamObserver {
   /**
    *
    * @param {Object} param
-   * @param {Connection} connection
-   * @param {} param.moreFunction -
-   * @param {} param.discardFunction -
-   * @param {} param.batchSize -
+   * @param {Connection} param.connection
+   * @param {boolean} param.reactive
+   * @param {function(connection: Connection, stmtId: number|Integer, n: number|Integer, observer: StreamObserver)} param.moreFunction -
+   * @param {function(connection: Connection, stmtId: number|Integer, observer: StreamObserver)} param.discardFunction -
+   * @param {number|Integer} param.batchSize -
    * @param {function(err: Error): Promise|void} param.beforeError -
    * @param {function(err: Error): Promise|void} param.afterError -
    * @param {function(keys: string[]): Promise|void} param.beforeKeys -
@@ -58,6 +60,7 @@ class ResultStreamObserver extends StreamObserver {
    */
   constructor ({
     connection,
+    reactive = false,
     moreFunction,
     discardFunction,
     batchSize = DefaultBatchSize,
@@ -71,6 +74,8 @@ class ResultStreamObserver extends StreamObserver {
     super()
 
     this._connection = connection
+    this._reactive = reactive
+    this._streaming = false
 
     this._fieldKeys = null
     this._fieldLookup = null
@@ -105,7 +110,11 @@ class ResultStreamObserver extends StreamObserver {
   onNext (rawRecord) {
     let record = new Record(this._fieldKeys, rawRecord, this._fieldLookup)
     if (this._observers.some(o => o.onNext)) {
-      this._observers.forEach(o => (o.onNext ? o.onNext(record) : {}))
+      this._observers.forEach(o => {
+        if (o.onNext) {
+          o.onNext(record)
+        }
+      })
     } else {
       this._queuedRecords.push(record)
     }
@@ -149,14 +158,20 @@ class ResultStreamObserver extends StreamObserver {
         this._head = this._fieldKeys
 
         if (this._observers.some(o => o.onKeys)) {
-          this._observers.forEach(o =>
-            o.onKeys ? o.onKeys(this._fieldKeys) : {}
-          )
+          this._observers.forEach(o => {
+            if (o.onKeys) {
+              o.onKeys(this._fieldKeys)
+            }
+          })
         }
 
         if (this._afterKeys) {
           this._afterKeys(this._fieldKeys)
         }
+
+        if (this._reactive) {
+          this._handleStreaming()
+        }
       }
 
       if (beforeHandlerResult) {
@@ -165,22 +180,13 @@ class ResultStreamObserver extends StreamObserver {
         continuation()
       }
     } else {
+      this._streaming = false
+
       if (meta.has_more) {
         // We've consumed current batch and server notified us that there're more
         // records to stream. Let's invoke more or discard function based on whether
         // the user wants to discard streaming or not
-        if (this._discard) {
-          this._discardFunction({
-            connection: this._connection,
-            statementId: this._statementId
-          })
-        } else {
-          this._moreFunction({
-            connection: this._connection,
-            statementId: this._statementId,
-            n: this._batchSize
-          })
-        }
+        this._handleStreaming()
 
         delete meta.has_more
       } else {
@@ -200,9 +206,11 @@ class ResultStreamObserver extends StreamObserver {
           this._tail = completionMetadata
 
           if (this._observers.some(o => o.onCompleted)) {
-            this._observers.forEach(o =>
-              o.onCompleted ? o.onCompleted(completionMetadata) : {}
-            )
+            this._observers.forEach(o => {
+              if (o.onCompleted) {
+                o.onCompleted(completionMetadata)
+              }
+            })
           }
 
           if (this._afterComplete) {
@@ -219,6 +227,28 @@ class ResultStreamObserver extends StreamObserver {
     }
   }
 
+  _handleStreaming () {
+    if (
+      this._reactive &&
+      this._head &&
+      this._observers.some(o => o.onNext || o.onCompleted) &&
+      !this._streaming
+    ) {
+      this._streaming = true
+
+      if (this._discard) {
+        this._discardFunction(this._connection, this._statementId, this)
+      } else {
+        this._moreFunction(
+          this._connection,
+          this._statementId,
+          this._batchSize,
+          this
+        )
+      }
+    }
+  }
+
   _storeMetadataForCompletion (meta) {
     const keys = Object.keys(meta)
     let index = keys.length
@@ -282,9 +312,11 @@ class ResultStreamObserver extends StreamObserver {
 
     const continuation = () => {
       if (this._observers.some(o => o.onError)) {
-        this._observers.forEach(o =>
-          o.onError ? o.onError(error) : console.log(error)
-        )
+        this._observers.forEach(o => {
+          if (o.onError) {
+            o.onError(error)
+          }
+        })
       }
 
       if (this._afterError) {
@@ -324,6 +356,10 @@ class ResultStreamObserver extends StreamObserver {
       observer.onCompleted(this._tail)
     }
     this._observers.push(observer)
+
+    if (this._reactive) {
+      this._handleStreaming()
+    }
   }
 
   hasFailed () {