Add sub milliseconds behaviour to Rx

Author: bigmontz

packages/neo4j-driver/src/driver.js

@@ -72,9 +72,11 @@ class Driver extends CoreDriver {
         reactive: false,
         fetchSize: validateFetchSizeValue(fetchSize, this._config.fetchSize),
         bookmarkManager,
-        notificationFilter
+        notificationFilter,
+        log: this._log
       }),
-      config: this._config
+      config: this._config,
+      log: this._log
     })
   }
 }

packages/neo4j-driver/src/session-rx.js

@@ -41,9 +41,10 @@ export default class RxSession {
    * @param {Object} param - Object parameter
    * @param {Session} param.session - The underlying session instance to relay requests
    */
-  constructor ({ session, config } = {}) {
+  constructor ({ session, config, log } = {}) {
     this._session = session
     this._retryLogic = _createRetryLogic(config)
+    this._log = log
   }
 
   /**
@@ -200,7 +201,7 @@ export default class RxSession {
   _beginTransaction (accessMode, transactionConfig) {
     let txConfig = TxConfig.empty()
     if (transactionConfig) {
-      txConfig = new TxConfig(transactionConfig)
+      txConfig = new TxConfig(transactionConfig, this._log)
     }
 
     return new Observable(observer => {

packages/neo4j-driver/test/driver.test.js

@@ -174,6 +174,18 @@ describe('#unit driver', () => {
         expect(session._session._bookmarkManager).toEqual(configuredBookmarkManager)
       })
     })
+
+    it('should redirect logger to session', () => {
+      driver = neo4j.driver(
+        `neo4j+ssc://${sharedNeo4j.hostname}`,
+        sharedNeo4j.authToken
+      )
+
+      const session = driver.rxSession()
+
+      expect(session._log).toBe(driver._log)
+      expect(session._session._log).toBe(driver._log)
+    })
   })
 })
 

packages/neo4j-driver/test/rx/session.test.js

@@ -19,7 +19,7 @@
 
 import { Notification, throwError } from 'rxjs'
 import { map, materialize, toArray, concatWith } from 'rxjs/operators'
-import neo4j from '../../src'
+import neo4j, { ResultSummary, int } from '../../src'
 import RxSession from '../../src/session-rx'
 import sharedNeo4j from '../internal/shared-neo4j'
 import {
@@ -31,7 +31,7 @@ import {
 } from 'neo4j-driver-core'
 
 const { SERVICE_UNAVAILABLE, SESSION_EXPIRED } = error
-const { bookmarks } = internal
+const { bookmarks, logger } = internal
 
 describe('#integration rx-session', () => {
   let driver
@@ -442,6 +442,111 @@ describe('#unit rx-session', () => {
     })
   })
 
+  describe('.run()', () => {
+    it('should redirect run to the underline session', () => {
+      const capture = []
+      const _session = {
+        run: (...args) => {
+          capture.push(args)
+          const summary = new ResultSummary(args[0], args[1], {}, 5.3)
+          return {
+            async * [Symbol.asyncIterator] () {
+              return summary
+            },
+            async summary () {
+              return summary
+            }
+          }
+        }
+      }
+      const session = new RxSession({ session: _session })
+      const query = 'the query'
+      const params = {}
+      const txConfig = { timeout: 1000 }
+
+      session.run(query, params, txConfig).records().subscribe()
+
+      expect(capture).toEqual([[query, params, txConfig]])
+    })
+  })
+
+  describe('._beginTransaction()', () => {
+    it('should round up sub milliseconds transaction timeouts', () => {
+      const capture = []
+      const _session = {
+        _beginTransaction: async (...args) => {
+          capture.push(args)
+          return {}
+        }
+      }
+
+      const session = new RxSession({
+        session: _session
+      })
+
+      const accessMode = 'READ'
+      const timeout = 0.2
+
+      session._beginTransaction(accessMode, { timeout })
+        .subscribe()
+
+      expect(capture[0][1].timeout).toEqual(int(1))
+    })
+
+    it('should log a warning for timeout configurations with sub milliseconds', () => {
+      const capture = []
+      const loggerFunction = (...args) => capture.push(args)
+      const log = new logger.Logger('debug', loggerFunction)
+
+      const _session = {
+        _beginTransaction: async (...args) => {
+          return {}
+        }
+      }
+
+      const session = new RxSession({
+        session: _session,
+        log
+      })
+
+      const accessMode = 'READ'
+      const timeout = 0.2
+
+      session._beginTransaction(accessMode, { timeout })
+        .subscribe()
+
+      expect(capture[0]).toEqual([
+        'info',
+        `Transaction timeout expected to be an integer, got: ${timeout}. The value will be round up.`
+      ])
+    })
+
+    it('should not log a warning for timeout configurations without sub milliseconds', () => {
+      const capture = []
+      const loggerFunction = (...args) => capture.push(args)
+      const log = new logger.Logger('debug', loggerFunction)
+
+      const _session = {
+        _beginTransaction: async (...args) => {
+          return {}
+        }
+      }
+
+      const session = new RxSession({
+        session: _session,
+        log
+      })
+
+      const accessMode = 'READ'
+      const timeout = 1
+
+      session._beginTransaction(accessMode, { timeout })
+        .subscribe()
+
+      expect(capture.length).toEqual(0)
+    })
+  })
+
   function newSession (lastBookmarks = bookmarks.Bookmarks.empty()) {
     const connectionProvider = new ConnectionProvider()
     connectionProvider.acquireConnection = () => Promise.resolve(null)