Add notificationFilter to the bolt-messages in Bolt 5.2

Author: bigmontz

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

@@ -0,0 +1,172 @@
+/**
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import BoltProtocolV5x1 from './bolt-protocol-v5x1'
+
+import transformersFactories from './bolt-protocol-v5x2.transformer'
+import Transformer from './transformer'
+import RequestMessage from './request-message'
+import { LoginObserver, ResultStreamObserver } from './stream-observers'
+
+import { internal } from 'neo4j-driver-core'
+
+const {
+  constants: { BOLT_PROTOCOL_V5_2, FETCH_ALL }
+} = internal
+
+export default class BoltProtocol extends BoltProtocolV5x1 {
+  get version () {
+    return BOLT_PROTOCOL_V5_2
+  }
+
+  get transformer () {
+    if (this._transformer === undefined) {
+      this._transformer = new Transformer(Object.values(transformersFactories).map(create => create(this._config, this._log)))
+    }
+    return this._transformer
+  }
+
+  get supportsReAuth () {
+    return true
+  }
+
+  /**
+   * Initialize a connection with the server
+   *
+   * @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 filters.
+   * @param {function(error)} args.onError On error callback
+   * @param {function(onComplete)} args.onComplete On complete callback
+   * @returns {LoginObserver} The Login observer
+   */
+  initialize ({ userAgent, authToken, notificationFilter, onError, onComplete } = {}) {
+    const state = {}
+    const observer = new LoginObserver({
+      onError: error => this._onLoginError(error, onError),
+      onCompleted: metadata => {
+        state.metadata = metadata
+        return this._onLoginCompleted(metadata)
+      }
+    })
+
+    this.write(
+      RequestMessage.hello5x2(userAgent, notificationFilter, this._serversideRouting),
+      observer,
+      false
+    )
+
+    return this.logon({
+      authToken,
+      onComplete: metadata => onComplete({ ...metadata, ...state.metadata }),
+      onError,
+      flush: true
+    })
+  }
+
+  beginTransaction ({
+    bookmarks,
+    txConfig,
+    database,
+    mode,
+    impersonatedUser,
+    notificationFilter,
+    beforeError,
+    afterError,
+    beforeComplete,
+    afterComplete
+  } = {}) {
+    const observer = new ResultStreamObserver({
+      server: this._server,
+      beforeError,
+      afterError,
+      beforeComplete,
+      afterComplete
+    })
+    observer.prepareToHandleSingleResponse()
+
+    this.write(
+      RequestMessage.begin({ bookmarks, txConfig, database, mode, impersonatedUser, notificationFilter }),
+      observer,
+      true
+    )
+
+    return observer
+  }
+
+  run (
+    query,
+    parameters,
+    {
+      bookmarks,
+      txConfig,
+      database,
+      mode,
+      impersonatedUser,
+      notificationFilter,
+      beforeKeys,
+      afterKeys,
+      beforeError,
+      afterError,
+      beforeComplete,
+      afterComplete,
+      flush = true,
+      reactive = false,
+      fetchSize = FETCH_ALL,
+      highRecordWatermark = Number.MAX_VALUE,
+      lowRecordWatermark = Number.MAX_VALUE
+    } = {}
+  ) {
+    const observer = new ResultStreamObserver({
+      server: this._server,
+      reactive: reactive,
+      fetchSize: fetchSize,
+      moreFunction: this._requestMore.bind(this),
+      discardFunction: this._requestDiscard.bind(this),
+      beforeKeys,
+      afterKeys,
+      beforeError,
+      afterError,
+      beforeComplete,
+      afterComplete,
+      highRecordWatermark,
+      lowRecordWatermark
+    })
+
+    const flushRun = reactive
+    this.write(
+      RequestMessage.runWithMetadata(query, parameters, {
+        bookmarks,
+        txConfig,
+        database,
+        mode,
+        impersonatedUser,
+        notificationFilter
+      }),
+      observer,
+      flushRun && flush
+    )
+
+    if (!reactive) {
+      this.write(RequestMessage.pull({ n: fetchSize }), observer, flush)
+    }
+
+    return observer
+  }
+}

packages/bolt-connection/src/bolt/bolt-protocol-v5x2.transformer.js

@@ -0,0 +1,24 @@
+/**
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import v5x0 from './bolt-protocol-v5x0.transformer'
+
+export default {
+  ...v5x0
+}

packages/bolt-connection/src/bolt/create.js

@@ -28,6 +28,7 @@ import BoltProtocolV4x3 from './bolt-protocol-v4x3'
 import BoltProtocolV4x4 from './bolt-protocol-v4x4'
 import BoltProtocolV5x0 from './bolt-protocol-v5x0'
 import BoltProtocolV5x1 from './bolt-protocol-v5x1'
+import BoltProtocolV5x2 from './bolt-protocol-v5x2'
 // eslint-disable-next-line no-unused-vars
 import { Chunker, Dechunker } from '../channel'
 import ResponseHandler from './response-handler'
@@ -202,6 +203,16 @@ function createProtocol (
         onProtocolError,
         serversideRouting
       )
+    case 5.2:
+      return new BoltProtocolV5x2(
+        server,
+        chunker,
+        packingConfig,
+        createResponseHandler,
+        log,
+        onProtocolError,
+        serversideRouting
+      )
     default:
       throw newError('Unknown Bolt protocol version: ' + version)
   }

packages/bolt-connection/src/bolt/handshake.js

@@ -76,7 +76,7 @@ function parseNegotiatedResponse (buffer) {
  */
 function newHandshakeBuffer () {
   return createHandshakeMessage([
-    [version(5, 1), version(5, 0)],
+    [version(5, 2), version(5, 0)],
     [version(4, 4), version(4, 2)],
     version(4, 1),
     version(3, 0)

packages/bolt-connection/src/bolt/request-message.js

@@ -160,6 +160,37 @@ export default class RequestMessage {
     )
   }
 
+  /**
+   * Create a new HELLO message.
+   * @param {string} userAgent the user agent.
+   * @param {NotificationFilter} notificationFilter the notification filter configured
+   * @param {Object} routing server side routing, set to routing context to turn on server side routing (> 4.1)
+   * @return {RequestMessage} new HELLO message.
+   */
+  static hello5x2 (userAgent, notificationFilter = null, routing = null) {
+    const metadata = { user_agent: userAgent }
+
+    if (notificationFilter) {
+      if (notificationFilter.minimumSeverityLevel) {
+        metadata.notifications_minimum_severity = notificationFilter.minimumSeverityLevel
+      }
+
+      if (notificationFilter.disabledCategories) {
+        metadata.notifications_disabled_categories = notificationFilter.disabledCategories
+      }
+    }
+
+    if (routing) {
+      metadata.routing = routing
+    }
+
+    return new RequestMessage(
+      HELLO,
+      [metadata],
+      () => `HELLO ${json.stringify(metadata)}`
+    )
+  }
+
   /**
    * Create a new LOGON message.
    *
@@ -194,10 +225,11 @@ export default class RequestMessage {
    * @param {string} database the database name.
    * @param {string} mode the access mode.
    * @param {string} impersonatedUser the impersonated user.
+   * @param {NotificationFilter} notificationFilter the notification filter
    * @return {RequestMessage} new BEGIN message.
    */
-  static begin ({ bookmarks, txConfig, database, mode, impersonatedUser } = {}) {
-    const metadata = buildTxMetadata(bookmarks, txConfig, database, mode, impersonatedUser)
+  static begin ({ bookmarks, txConfig, database, mode, impersonatedUser, notificationFilter } = {}) {
+    const metadata = buildTxMetadata(bookmarks, txConfig, database, mode, impersonatedUser, notificationFilter)
     return new RequestMessage(
       BEGIN,
       [metadata],
@@ -235,9 +267,9 @@ export default class RequestMessage {
   static runWithMetadata (
     query,
     parameters,
-    { bookmarks, txConfig, database, mode, impersonatedUser } = {}
+    { bookmarks, txConfig, database, mode, impersonatedUser, notificationFilter } = {}
   ) {
-    const metadata = buildTxMetadata(bookmarks, txConfig, database, mode, impersonatedUser)
+    const metadata = buildTxMetadata(bookmarks, txConfig, database, mode, impersonatedUser, notificationFilter)
     return new RequestMessage(
       RUN,
       [query, parameters, metadata],
@@ -348,9 +380,10 @@ export default class RequestMessage {
  * @param {string} database the database name.
  * @param {string} mode the access mode.
  * @param {string} impersonatedUser the impersonated user mode.
+ * @param {notificationFilter} notificationFilter the notification filter
  * @return {Object} a metadata object.
  */
-function buildTxMetadata (bookmarks, txConfig, database, mode, impersonatedUser) {
+function buildTxMetadata (bookmarks, txConfig, database, mode, impersonatedUser, notificationFilter) {
   const metadata = {}
   if (!bookmarks.isEmpty()) {
     metadata.bookmarks = bookmarks.values()
@@ -370,6 +403,15 @@ function buildTxMetadata (bookmarks, txConfig, database, mode, impersonatedUser)
   if (mode === ACCESS_MODE_READ) {
     metadata.mode = READ_MODE
   }
+  if (notificationFilter) {
+    if (notificationFilter.minimumSeverityLevel) {
+      metadata.notifications_minimum_severity = notificationFilter.minimumSeverityLevel
+    }
+
+    if (notificationFilter.disabledCategories) {
+      metadata.notifications_disabled_categories = notificationFilter.disabledCategories
+    }
+  }
   return metadata
 }
 

packages/bolt-connection/src/connection/connection-channel.js

@@ -86,6 +86,7 @@ export function createChannelConnection (
         config.disableLosslessIntegers,
         serversideRouting,
         chunker,
+        config.notificationFilter,
         createProtocol
       )
 
@@ -119,6 +120,7 @@ export default class ChannelConnection extends Connection {
     disableLosslessIntegers = false,
     serversideRouting = null,
     chunker, // to be removed,
+    notificationFilter,
     protocolSupplier
   ) {
     super(errorHandler)
@@ -134,6 +136,7 @@ export default class ChannelConnection extends Connection {
     this._chunker = chunker
     this._log = createConnectionLogger(this, log)
     this._serversideRouting = serversideRouting
+    this._notificationFilter = notificationFilter
 
     // connection from the database, returned in response for HELLO message and might not be available
     this._dbConnectionId = null
@@ -187,6 +190,7 @@ export default class ChannelConnection extends Connection {
       this._protocol.initialize({
         userAgent,
         authToken,
+        notificationFilter: this._notificationFilter,
         onError: err => reject(err),
         onComplete: metadata => {
           if (metadata) {