feat: sql filters, properties filter, application properties filter

Author: magne

src/consumer.ts

@@ -8,15 +8,19 @@ import {
   EventContext,
   Message,
   Dictionary,
-  filter,
+  types,
+  Typed,
+  MessageProperties,
 } from "rhea"
 import {
   Offset,
   SourceFilter,
   STREAM_FILTER_MATCH_UNFILTERED,
   STREAM_FILTER_SPEC,
   STREAM_OFFSET_SPEC,
-  STREAM_SQL_FILTER,
+  STREAM_FILTER_SQL,
+  STREAM_FILTER_MESSAGE_PROPERTIES,
+  STREAM_FILTER_APPLICATION_PROPERTIES,
 } from "./utils.js"
 import { openLink } from "./rhea_wrapper.js"
 import { createConsumerAddressFrom } from "./message.js"
@@ -29,6 +33,8 @@ export type StreamOptions = {
   name: string
   offset?: Offset
   filterValues?: string[]
+  messagePropertiesFilter?: MessageProperties
+  applicationPropertiesFilter?: Dictionary<string>
   sqlFilter?: string
   matchUnfiltered?: boolean
 }
@@ -128,21 +134,27 @@ function createConsumerFilterFrom(params: CreateConsumerParams): SourceFilter |
     throw new Error("At least one between offset and filterValues must be set when using filtering")
   }
 
-  const filters: Dictionary<string | bigint | boolean | string[] | any> = {}
+  const filters: Dictionary<string | bigint | boolean | string[] | Typed> = {}
   if (params.stream.offset) {
     filters[STREAM_OFFSET_SPEC] = params.stream.offset.toValue()
   }
   if (params.stream.filterValues) {
     filters[STREAM_FILTER_SPEC] = params.stream.filterValues
   }
   if (params.stream.sqlFilter) {
-    console.log(filter.selector(`Described(${String(Symbol("amqp:sql-filter"))}, ${params.stream.sqlFilter})`))
-    const { descriptor, value } = filter.selector(params.stream.sqlFilter)["jms-selector"]
-    filters["sql-filter"] = { descriptor: descriptor.value, value }
+    filters[STREAM_FILTER_SQL] = types.wrap_described(params.stream.sqlFilter, 0x120)
   }
   if (params.stream.matchUnfiltered) {
     filters[STREAM_FILTER_MATCH_UNFILTERED] = params.stream.matchUnfiltered
   }
+  if (params.stream.messagePropertiesFilter) {
+    const symbolicMap = types.wrap_symbolic_map(params.stream.messagePropertiesFilter)
+    filters[STREAM_FILTER_MESSAGE_PROPERTIES] = types.wrap_described(symbolicMap, 0x173)
+  }
+  if (params.stream.applicationPropertiesFilter) {
+    const map = types.wrap_map(params.stream.applicationPropertiesFilter)
+    filters[STREAM_FILTER_APPLICATION_PROPERTIES] = types.wrap_described(map, 0x174)
+  }
 
   return filters
 }

src/utils.ts

@@ -1,4 +1,4 @@
-import { Dictionary, Message } from "rhea"
+import { Dictionary, Message, Typed } from "rhea"
 import { QueueType } from "./queue.js"
 
 export enum AmqpResponseCodes {
@@ -20,12 +20,14 @@ export const DURABLE = 1
 export const AUTO_DELETE = 1
 export const EXCLUSIVE = 1
 
-export const STREAM_FILTER_SPEC = "rabbitmq:stream-filter"
 export const STREAM_OFFSET_SPEC = "rabbitmq:stream-offset-spec"
+export const STREAM_FILTER_SPEC = "rabbitmq:stream-filter"
 export const STREAM_FILTER_MATCH_UNFILTERED = "rabbitmq:stream-match-unfiltered"
-export const STREAM_SQL_FILTER = "amqp:sql-filter"
+export const STREAM_FILTER_MESSAGE_PROPERTIES = "properties-filter"
+export const STREAM_FILTER_APPLICATION_PROPERTIES = "application-properties-filter"
+export const STREAM_FILTER_SQL = "sql-filter"
 
-export type SourceFilter = Dictionary<string | bigint | boolean | string[]>
+export type SourceFilter = Dictionary<string | bigint | boolean | string[] | Typed>
 
 export type Result<T, K> = OkResult<T> | ErrorResult<K>
 

test/e2e/consumer.test.ts

@@ -202,7 +202,81 @@ describe("Consumer", () => {
     })
   })
 
-  test("consumer can handle message on stream with sql filters", async () => {
+  test("consumer can handle message on stream with filter on message properties", async () => {
+    const publisher = await connection.createPublisher({ queue: { name: streamName } })
+    const filteredMessage = createAmqpMessage({
+      body: "my body",
+      message_properties: {
+        subject: "foo",
+      },
+    })
+    const discardedMessage = createAmqpMessage({
+      body: "discard me",
+      message_properties: {
+        subject: "bar",
+      },
+    })
+    await publisher.publish(filteredMessage)
+    await publisher.publish(discardedMessage)
+    let received: number = 0
+
+    const consumer = await connection.createConsumer({
+      stream: {
+        name: streamName,
+        offset: Offset.first(),
+        matchUnfiltered: false,
+        messagePropertiesFilter: { subject: "foo" },
+      },
+      messageHandler: (context) => {
+        received++
+        context.accept()
+      },
+    })
+    consumer.start()
+
+    await eventually(() => {
+      expect(received).to.be.eql(1)
+    })
+  })
+
+  test("consumer can handle message on stream with filter on application properties", async () => {
+    const publisher = await connection.createPublisher({ queue: { name: streamName } })
+    const filteredMessage = createAmqpMessage({
+      body: "my body",
+      application_properties: {
+        test: "foo",
+      },
+    })
+    const discardedMessage = createAmqpMessage({
+      body: "discard me",
+      application_properties: {
+        test: "bar",
+      },
+    })
+    await publisher.publish(filteredMessage)
+    await publisher.publish(discardedMessage)
+    let received: number = 0
+
+    const consumer = await connection.createConsumer({
+      stream: {
+        name: streamName,
+        offset: Offset.first(),
+        matchUnfiltered: false,
+        applicationPropertiesFilter: { test: "foo" },
+      },
+      messageHandler: (context) => {
+        received++
+        context.accept()
+      },
+    })
+    consumer.start()
+
+    await eventually(() => {
+      expect(received).to.be.eql(1)
+    })
+  })
+
+  test("consumer can handle message on stream with SQL filters on message properties", async () => {
     const publisher = await connection.createPublisher({ queue: { name: streamName } })
     const filteredMessage = createAmqpMessage({
       body: "my body",
@@ -225,12 +299,49 @@ describe("Consumer", () => {
         name: streamName,
         offset: Offset.first(),
         matchUnfiltered: false,
-        sqlFilter: "properties.subject = '123'",
+        sqlFilter: "properties.subject = 'foo'",
       },
       messageHandler: (context, message) => {
-        console.log("message", message.subject)
         if (message.subject && message.subject == "foo") {
-          console.log("hello ")
+          received = message.body
+        }
+        context.accept()
+      },
+    })
+    consumer.start()
+
+    await eventually(() => {
+      expect(received).to.be.eql("my body")
+    })
+  })
+
+  test("consumer can handle message on stream with SQL filters on message application properties", async () => {
+    const publisher = await connection.createPublisher({ queue: { name: streamName } })
+    const filteredMessage = createAmqpMessage({
+      body: "my body",
+      application_properties: {
+        test: "foo",
+      },
+    })
+    const discardedMessage = createAmqpMessage({
+      body: "discard me",
+      application_properties: {
+        test: "bar",
+      },
+    })
+    await publisher.publish(filteredMessage)
+    await publisher.publish(discardedMessage)
+    let received: string = ""
+
+    const consumer = await connection.createConsumer({
+      stream: {
+        name: streamName,
+        offset: Offset.first(),
+        matchUnfiltered: false,
+        sqlFilter: "application_properties.test = 'foo'",
+      },
+      messageHandler: (context, message) => {
+        if (message.application_properties && message.application_properties.test == "foo") {
           received = message.body
         }
         context.accept()