Database Notification Improvements

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

@@ -16,7 +16,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { newError } from 'neo4j-driver-core'
+import { newError, json } from 'neo4j-driver-core'
 // eslint-disable-next-line no-unused-vars
 import { ResultStreamObserver } from './stream-observers'
 
@@ -79,4 +79,24 @@ function assertImpersonatedUserIsEmpty (impersonatedUser, onProtocolError = () =
   }
 }
 
-export { assertDatabaseIsEmpty, assertTxConfigIsEmpty, assertImpersonatedUserIsEmpty }
+/**
+ * Asserts that the passed-in notificationFilter is empty
+ * @param {NotificationFilter} notificationFilter
+ * @param {function (err:Error)} onProtocolError Called when it does have notificationFilter user set
+ * @param {any} observer
+ */
+function assertNotificationFilterIsEmpty (notificationFilter, onProtocolError = () => {}, observer) {
+  if (notificationFilter !== undefined) {
+    const error = newError(
+      'Driver is connected to a database that does not support user notification filters. ' +
+        'Please upgrade to Neo4j 5.7.0 or later in order to use this functionality. ' +
+        `Trying to set notifications to ${json.stringify(notificationFilter)}.`
+    )
+    // unsupported API was used, consider this a fatal error for the current connection
+    onProtocolError(error.message)
+    observer.onError(error)
+    throw error
+  }
+}
+
+export { assertDatabaseIsEmpty, assertTxConfigIsEmpty, assertImpersonatedUserIsEmpty, assertNotificationFilterIsEmpty }

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

@@ -19,7 +19,8 @@
 import {
   assertDatabaseIsEmpty,
   assertTxConfigIsEmpty,
-  assertImpersonatedUserIsEmpty
+  assertImpersonatedUserIsEmpty,
+  assertNotificationFilterIsEmpty
 } from './bolt-protocol-util'
 // eslint-disable-next-line no-unused-vars
 import { Chunker } from '../channel'
@@ -169,16 +170,20 @@ export default class BoltProtocol {
    * @param {Object} param
    * @param {string} param.userAgent the user agent.
    * @param {Object} param.authToken the authentication token.
+   * @param {NotificationFilter} param.notificationFilter the notification filter.
    * @param {function(err: Error)} param.onError the callback to invoke on error.
    * @param {function()} param.onComplete the callback to invoke on completion.
    * @returns {StreamObserver} the stream observer that monitors the corresponding server response.
    */
-  initialize ({ userAgent, authToken, onError, onComplete } = {}) {
+  initialize ({ userAgent, authToken, notificationFilter, onError, onComplete } = {}) {
     const observer = new LoginObserver({
       onError: error => this._onLoginError(error, onError),
       onCompleted: metadata => this._onLoginCompleted(metadata, onComplete)
     })
 
+    // passing notification filter on this protocol version throws an error
+    assertNotificationFilterIsEmpty(notificationFilter, this._onProtocolError, observer)
+
     this.write(RequestMessage.init(userAgent, authToken), observer, true)
 
     return observer
@@ -254,6 +259,7 @@ export default class BoltProtocol {
    * @param {string} param.database the target database name.
    * @param {string} param.mode the access mode.
    * @param {string} param.impersonatedUser the impersonated user
+   * @param {NotificationFilter} param.notificationFilter the notification filter.
    * @param {function(err: Error)} param.beforeError the callback to invoke before handling the error.
    * @param {function(err: Error)} param.afterError the callback to invoke after handling the error.
    * @param {function()} param.beforeComplete the callback to invoke before handling the completion.
@@ -266,6 +272,7 @@ export default class BoltProtocol {
     database,
     mode,
     impersonatedUser,
+    notificationFilter,
     beforeError,
     afterError,
     beforeComplete,
@@ -280,6 +287,7 @@ export default class BoltProtocol {
         database,
         mode,
         impersonatedUser,
+        notificationFilter,
         beforeError,
         afterError,
         beforeComplete,
@@ -362,6 +370,7 @@ export default class BoltProtocol {
    * @param {TxConfig} param.txConfig the transaction configuration.
    * @param {string} param.database the target database name.
    * @param {string} param.impersonatedUser the impersonated user
+   * @param {NotificationFilter} param.notificationFilter the notification filter.
    * @param {string} param.mode the access mode.
    * @param {function(keys: string[])} param.beforeKeys the callback to invoke before handling the keys.
    * @param {function(keys: string[])} param.afterKeys the callback to invoke after handling the keys.
@@ -381,6 +390,7 @@ export default class BoltProtocol {
       database,
       mode,
       impersonatedUser,
+      notificationFilter,
       beforeKeys,
       afterKeys,
       beforeError,
@@ -410,6 +420,8 @@ export default class BoltProtocol {
     assertDatabaseIsEmpty(database, this._onProtocolError, observer)
     // passing impersonated user on this protocol version throws an error
     assertImpersonatedUserIsEmpty(impersonatedUser, this._onProtocolError, observer)
+    // passing notification filter on this protocol version throws an error
+    assertNotificationFilterIsEmpty(notificationFilter, this._onProtocolError, observer)
 
     this.write(RequestMessage.run(query, parameters), observer, false)
     this.write(RequestMessage.pullAll(), observer, flush)

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

@@ -18,7 +18,7 @@
  */
 import BoltProtocolV2 from './bolt-protocol-v2'
 import RequestMessage from './request-message'
-import { assertDatabaseIsEmpty, assertImpersonatedUserIsEmpty } from './bolt-protocol-util'
+import { assertDatabaseIsEmpty, assertImpersonatedUserIsEmpty, assertNotificationFilterIsEmpty } from './bolt-protocol-util'
 import {
   StreamObserver,
   LoginObserver,
@@ -69,12 +69,15 @@ export default class BoltProtocol extends BoltProtocolV2 {
     return metadata
   }
 
-  initialize ({ userAgent, authToken, onError, onComplete } = {}) {
+  initialize ({ userAgent, authToken, notificationFilter, onError, onComplete } = {}) {
     const observer = new LoginObserver({
       onError: error => this._onLoginError(error, onError),
       onCompleted: metadata => this._onLoginCompleted(metadata, authToken, onComplete)
     })
 
+    // passing notification filter on this protocol version throws an error
+    assertNotificationFilterIsEmpty(notificationFilter, this._onProtocolError, observer)
+
     this.write(RequestMessage.hello(userAgent, authToken), observer, true)
 
     return observer
@@ -89,6 +92,7 @@ export default class BoltProtocol extends BoltProtocolV2 {
     txConfig,
     database,
     impersonatedUser,
+    notificationFilter,
     mode,
     beforeError,
     afterError,
@@ -108,6 +112,8 @@ export default class BoltProtocol extends BoltProtocolV2 {
     assertDatabaseIsEmpty(database, this._onProtocolError, observer)
     // passing impersonated user on this protocol version throws an error
     assertImpersonatedUserIsEmpty(impersonatedUser, this._onProtocolError, observer)
+    // passing notification filter on this protocol version throws an error
+    assertNotificationFilterIsEmpty(notificationFilter, this._onProtocolError, observer)
 
     this.write(
       RequestMessage.begin({ bookmarks, txConfig, mode }),
@@ -166,6 +172,7 @@ export default class BoltProtocol extends BoltProtocolV2 {
       txConfig,
       database,
       impersonatedUser,
+      notificationFilter,
       mode,
       beforeKeys,
       afterKeys,
@@ -194,6 +201,8 @@ export default class BoltProtocol extends BoltProtocolV2 {
     assertDatabaseIsEmpty(database, this._onProtocolError, observer)
     // passing impersonated user on this protocol version throws an error
     assertImpersonatedUserIsEmpty(impersonatedUser, this._onProtocolError, observer)
+    // passing notification filter on this protocol version throws an error
+    assertNotificationFilterIsEmpty(notificationFilter, this._onProtocolError, observer)
 
     this.write(
       RequestMessage.runWithMetadata(query, parameters, {

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

@@ -18,7 +18,7 @@
  */
 import BoltProtocolV3 from './bolt-protocol-v3'
 import RequestMessage from './request-message'
-import { assertImpersonatedUserIsEmpty } from './bolt-protocol-util'
+import { assertImpersonatedUserIsEmpty, assertNotificationFilterIsEmpty } from './bolt-protocol-util'
 import {
   ResultStreamObserver,
   ProcedureRouteObserver
@@ -56,6 +56,7 @@ export default class BoltProtocol extends BoltProtocolV3 {
     txConfig,
     database,
     impersonatedUser,
+    notificationFilter,
     mode,
     beforeError,
     afterError,
@@ -73,6 +74,8 @@ export default class BoltProtocol extends BoltProtocolV3 {
 
     // passing impersonated user on this protocol version throws an error
     assertImpersonatedUserIsEmpty(impersonatedUser, this._onProtocolError, observer)
+    // passing notification filter on this protocol version throws an error
+    assertNotificationFilterIsEmpty(notificationFilter, this._onProtocolError, observer)
 
     this.write(
       RequestMessage.begin({ bookmarks, txConfig, database, mode }),
@@ -91,6 +94,7 @@ export default class BoltProtocol extends BoltProtocolV3 {
       txConfig,
       database,
       impersonatedUser,
+      notificationFilter,
       mode,
       beforeKeys,
       afterKeys,
@@ -123,6 +127,8 @@ export default class BoltProtocol extends BoltProtocolV3 {
 
     // passing impersonated user on this protocol version throws an error
     assertImpersonatedUserIsEmpty(impersonatedUser, this._onProtocolError, observer)
+    // passing notification filter on this protocol version throws an error
+    assertNotificationFilterIsEmpty(notificationFilter, this._onProtocolError, observer)
 
     const flushRun = reactive
     this.write(

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

@@ -20,6 +20,7 @@ import BoltProtocolV4 from './bolt-protocol-v4x0'
 import RequestMessage from './request-message'
 import { LoginObserver } from './stream-observers'
 import { internal } from 'neo4j-driver-core'
+import { assertNotificationFilterIsEmpty } from './bolt-protocol-util'
 
 import transformersFactories from './bolt-protocol-v4x1.transformer'
 import Transformer from './transformer'
@@ -72,12 +73,15 @@ export default class BoltProtocol extends BoltProtocolV4 {
     return this._transformer
   }
 
-  initialize ({ userAgent, authToken, onError, onComplete } = {}) {
+  initialize ({ userAgent, authToken, notificationFilter, onError, onComplete } = {}) {
     const observer = new LoginObserver({
       onError: error => this._onLoginError(error, onError),
       onCompleted: metadata => this._onLoginCompleted(metadata, authToken, onComplete)
     })
 
+    // passing notification filter on this protocol version throws an error
+    assertNotificationFilterIsEmpty(notificationFilter, this._onProtocolError, observer)
+
     this.write(
       RequestMessage.hello(userAgent, authToken, this._serversideRouting),
       observer,

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

@@ -19,6 +19,7 @@
 import BoltProtocolV42 from './bolt-protocol-v4x2'
 import RequestMessage from './request-message'
 import { LoginObserver, RouteObserver } from './stream-observers'
+import { assertNotificationFilterIsEmpty } from './bolt-protocol-util'
 
 import transformersFactories from './bolt-protocol-v4x3.transformer'
 import utcTransformersFactories from './bolt-protocol-v5x0.utc.transformer'
@@ -80,14 +81,15 @@ export default class BoltProtocol extends BoltProtocolV42 {
   /**
    * Initialize a connection with the server
    *
-   * @param {Object} param0 The params
-   * @param {string} param0.userAgent The user agent
-   * @param {any} param0.authToken The auth token
-   * @param {function(error)} param0.onError On error callback
-   * @param {function(onComplte)} param0.onComplete On complete callback
+   * @param {Object} args The params
+   * @param {string} args.userAgent The user agent
+   * @param {any} args.authToken The auth token
+   * @param {NotificationFilter} args.notificationFilter The notification filter.
+   * @param {function(error)} args.onError On error callback
+   * @param {function(onComplte)} args.onComplete On complete callback
    * @returns {LoginObserver} The Login observer
    */
-  initialize ({ userAgent, authToken, onError, onComplete } = {}) {
+  initialize ({ userAgent, authToken, notificationFilter, onError, onComplete } = {}) {
     const observer = new LoginObserver({
       onError: error => this._onLoginError(error, onError),
       onCompleted: metadata => {
@@ -98,6 +100,9 @@ export default class BoltProtocol extends BoltProtocolV42 {
       }
     })
 
+    // passing notification filter on this protocol version throws an error
+    assertNotificationFilterIsEmpty(notificationFilter, this._onProtocolError, observer)
+
     this.write(
       RequestMessage.hello(userAgent, authToken, this._serversideRouting, ['utc']),
       observer,

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

@@ -21,6 +21,7 @@ import BoltProtocolV43 from './bolt-protocol-v4x3'
 import { internal } from 'neo4j-driver-core'
 import RequestMessage from './request-message'
 import { RouteObserver, ResultStreamObserver } from './stream-observers'
+import { assertNotificationFilterIsEmpty } from './bolt-protocol-util'
 
 import transformersFactories from './bolt-protocol-v4x4.transformer'
 import utcTransformersFactories from './bolt-protocol-v5x0.utc.transformer'
@@ -87,6 +88,7 @@ export default class BoltProtocol extends BoltProtocolV43 {
       database,
       mode,
       impersonatedUser,
+      notificationFilter,
       beforeKeys,
       afterKeys,
       beforeError,
@@ -116,6 +118,9 @@ export default class BoltProtocol extends BoltProtocolV43 {
       lowRecordWatermark
     })
 
+    // passing notification filter on this protocol version throws an error
+    assertNotificationFilterIsEmpty(notificationFilter, this._onProtocolError, observer)
+
     const flushRun = reactive
     this.write(
       RequestMessage.runWithMetadata(query, parameters, {
@@ -142,6 +147,7 @@ export default class BoltProtocol extends BoltProtocolV43 {
     database,
     mode,
     impersonatedUser,
+    notificationFilter,
     beforeError,
     afterError,
     beforeComplete,
@@ -156,6 +162,9 @@ export default class BoltProtocol extends BoltProtocolV43 {
     })
     observer.prepareToHandleSingleResponse()
 
+    // passing notification filter on this protocol version throws an error
+    assertNotificationFilterIsEmpty(notificationFilter, this._onProtocolError, observer)
+
     this.write(
       RequestMessage.begin({ bookmarks, txConfig, database, mode, impersonatedUser }),
       observer,

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

@@ -18,6 +18,7 @@
  */
 import BoltProtocolV44 from './bolt-protocol-v4x4'
 
+import { assertNotificationFilterIsEmpty } from './bolt-protocol-util'
 import transformersFactories from './bolt-protocol-v5x0.transformer'
 import Transformer from './transformer'
 import RequestMessage from './request-message'
@@ -44,19 +45,23 @@ export default class BoltProtocol extends BoltProtocolV44 {
   /**
    * Initialize a connection with the server
    *
-   * @param {Object} param0 The params
-   * @param {string} param0.userAgent The user agent
-   * @param {any} param0.authToken The auth token
-   * @param {function(error)} param0.onError On error callback
-   * @param {function(onComplte)} param0.onComplete On complete callback
+   * @param {Object} args The params
+   * @param {string} args.userAgent The user agent
+   * @param {any} args.authToken The auth token
+   * @param {NotificationFilter} args.notificationFilter The notification filter.
+   * @param {function(error)} args.onError On error callback
+   * @param {function(onComplte)} args.onComplete On complete callback
    * @returns {LoginObserver} The Login observer
    */
-  initialize ({ userAgent, authToken, onError, onComplete } = {}) {
+  initialize ({ userAgent, authToken, notificationFilter, onError, onComplete } = {}) {
     const observer = new LoginObserver({
       onError: error => this._onLoginError(error, onError),
       onCompleted: metadata => this._onLoginCompleted(metadata, authToken, onComplete)
     })
 
+    // passing notification filter on this protocol version throws an error
+    assertNotificationFilterIsEmpty(notificationFilter, this._onProtocolError, observer)
+
     this.write(
       RequestMessage.hello(userAgent, authToken, this._serversideRouting),
       observer,