Adds filtering feature on consumer side (#172)

* Adds filtering feature with publish request v2 and deliver v2

* Npm check fix

* Revert of request interfaces

* Revert to standard time out for test

* PR review fixes

* Adds filtering on consuming side

* Rebase from filtering on publisher side

* Remove connection close on declare consumer with filter error

* Increase filtering test timeout

* Fix on consumer filtering

---------

Co-authored-by: magne <magnello@coders51.com>

Author: l4mby

src/client.ts

@@ -177,14 +177,23 @@ export class Client {
   public async declareConsumer(params: DeclareConsumerParams, handle: ConsumerFunc): Promise<Consumer> {
     const consumerId = this.incConsumerId()
     const properties: Record<string, string> = {}
-    const client = await this.getConnection(params.stream, false, params.connectionClosedListener)
-    const consumer = new StreamConsumer(addOffsetFilterToHandle(handle, params.offset), {
-      connection: client,
-      stream: params.stream,
-      consumerId,
-      consumerRef: params.consumerRef,
-      offset: params.offset,
-    })
+    const connection = await this.getConnection(params.stream, false, params.connectionClosedListener)
+
+    if (params.filter && !connection.isFilteringEnabled) {
+      throw new Error(`Broker does not support message filtering.`)
+    }
+
+    const consumer = new StreamConsumer(
+      addOffsetFilterToHandle(handle, params.offset),
+      {
+        connection,
+        stream: params.stream,
+        consumerId,
+        consumerRef: params.consumerRef,
+        offset: params.offset,
+      },
+      params.filter
+    )
     this.consumers.set(consumerId, consumer)
 
     if (params.singleActive && !params.consumerRef) {
@@ -194,6 +203,12 @@ export class Client {
       properties["single-active-consumer"] = "true"
       properties["name"] = params.consumerRef!
     }
+    if (params.filter) {
+      for (let i = 0; i < params.filter.values.length; i++) {
+        properties[`filter.${i}`] = params.filter.values[i]
+      }
+      properties["match-unfiltered"] = `${params.filter.matchUnfiltered}`
+    }
 
     const res = await this.connection.sendAndWait<SubscribeResponse>(
       new SubscribeRequest({ ...params, subscriptionId: consumerId, credit: 10, properties: properties })
@@ -441,6 +456,10 @@ export class Client {
       this.logger.debug(`on deliverV2 -> ${consumer.consumerRef}`)
       this.logger.debug(`response.messages.length: ${response.messages.length}`)
       await this.askForCredit({ credit: 1, subscriptionId: response.subscriptionId })
+      if (consumer.filter) {
+        response.messages.filter((x) => consumer.filter?.postFilterFunc(x)).map((x) => consumer.handle(x))
+        return
+      }
       response.messages.map((x) => consumer.handle(x))
     }
   }
@@ -610,12 +629,19 @@ export interface DeclareSuperStreamPublisherParams {
   routingStrategy?: RoutingStrategy
 }
 
+export interface ConsumerFilter {
+  values: string[]
+  postFilterFunc: (msg: Message) => boolean
+  matchUnfiltered: boolean
+}
+
 export interface DeclareConsumerParams {
   stream: string
   consumerRef?: string
   offset: Offset
   connectionClosedListener?: ConnectionClosedListener
   singleActive?: boolean
+  filter?: ConsumerFilter
 }
 
 export interface DeclareSuperStreamConsumerParams {

src/consumer.ts

@@ -1,3 +1,4 @@
+import { ConsumerFilter } from "./client"
 import { ConnectionInfo, Connection } from "./connection"
 import { ConnectionPool } from "./connection_pool"
 import { Message } from "./publisher"
@@ -29,7 +30,8 @@ export class StreamConsumer implements Consumer {
       consumerId: number
       consumerRef?: string
       offset: Offset
-    }
+    },
+    readonly filter?: ConsumerFilter
   ) {
     this.connection = params.connection
     this.stream = params.stream

test/e2e/filtering.test.ts

@@ -1,6 +1,6 @@
 import { expect } from "chai"
 import { randomUUID } from "crypto"
-import { Client } from "../../src"
+import { Client, Offset } from "../../src"
 import { createClient, createStreamName } from "../support/fake_data"
 import { Rabbit } from "../support/rabbit"
 import { eventually, username, password } from "../support/util"
@@ -47,4 +47,103 @@ describe("filtering", () => {
       expect((await rabbit.getQueueInfo(streamName)).messages).eql(3)
     }, 10000)
   }).timeout(10000)
+
+  it("published messages are filtered on the consumer side", async () => {
+    const filteredMsg: string[] = []
+    const publisher = await client.declarePublisher(
+      { stream: streamName, publisherRef: `my-publisher-${randomUUID()}` },
+      (msg) => msg.applicationProperties!["test"].toString()
+    )
+    const message1 = "test1"
+    const message2 = "test2"
+    const message3 = "test3"
+    const applicationProperties1 = { test: "A" }
+    const applicationProperties2 = { test: "B" }
+    const applicationProperties3 = { test: "C" }
+    await publisher.send(Buffer.from(message1), { applicationProperties: applicationProperties1 })
+    await publisher.send(Buffer.from(message2), { applicationProperties: applicationProperties2 })
+    await publisher.send(Buffer.from(message3), { applicationProperties: applicationProperties3 })
+
+    await client.declareConsumer(
+      {
+        stream: streamName,
+        offset: Offset.first(),
+        filter: {
+          values: ["A", "B"],
+          postFilterFunc: (msg) => msg.applicationProperties!["test"] === "A",
+          matchUnfiltered: true,
+        },
+      },
+      (msg) => filteredMsg.push(msg.content.toString("utf-8"))
+    )
+
+    await eventually(async () => {
+      expect(filteredMsg[0]).eql("test1")
+      expect(filteredMsg.length).eql(1)
+    }, 10000)
+  }).timeout(10000)
+
+  it("published messages are filtered on the server side keeping only the ones with filter value", async () => {
+    const filteredMsg: string[] = []
+    const publisher = await client.declarePublisher(
+      { stream: streamName, publisherRef: `my-publisher-${randomUUID()}` },
+      (msg) => (msg.applicationProperties ? msg.applicationProperties["test"].toString() : undefined)
+    )
+    const applicationProperties1 = { test: "A" }
+    const applicationProperties2 = { test: "B" }
+    for (let i = 0; i < 1000; i++)
+      await publisher.send(Buffer.from(`test${i + 1}`), { applicationProperties: applicationProperties1 })
+    for (let i = 0; i < 1000; i++)
+      await publisher.send(Buffer.from(`test${i + 1}`), { applicationProperties: applicationProperties2 })
+    for (let i = 0; i < 1000; i++) await publisher.send(Buffer.from(`test${i + 1}`))
+
+    await client.declareConsumer(
+      {
+        stream: streamName,
+        offset: Offset.first(),
+        filter: {
+          values: ["A", "B"],
+          postFilterFunc: (_msg) => true,
+          matchUnfiltered: false,
+        },
+      },
+      (msg) => filteredMsg.push(msg.content.toString("utf-8"))
+    )
+
+    await eventually(async () => {
+      expect(filteredMsg.length).eql(2000)
+    }, 10000)
+  }).timeout(10000)
+
+  it("published messages are filtered on the server side keeping even the ones with filter value", async () => {
+    const filteredMsg: string[] = []
+    const publisher = await client.declarePublisher(
+      { stream: streamName, publisherRef: `my-publisher-${randomUUID()}` },
+      (msg) => (msg.applicationProperties ? msg.applicationProperties["test"].toString() : undefined)
+    )
+    const applicationProperties1 = { test: "A" }
+    const applicationProperties2 = { test: "B" }
+    for (let i = 0; i < 1000; i++)
+      await publisher.send(Buffer.from(`test${i + 1}`), { applicationProperties: applicationProperties1 })
+    for (let i = 0; i < 1000; i++)
+      await publisher.send(Buffer.from(`test${i + 1}`), { applicationProperties: applicationProperties2 })
+    for (let i = 0; i < 1000; i++) await publisher.send(Buffer.from(`test${i + 1}`))
+
+    await client.declareConsumer(
+      {
+        stream: streamName,
+        offset: Offset.first(),
+        filter: {
+          values: ["A", "B"],
+          postFilterFunc: (_msg) => true,
+          matchUnfiltered: true,
+        },
+      },
+      (msg) => filteredMsg.push(msg.content.toString("utf-8"))
+    )
+
+    await eventually(async () => {
+      expect(filteredMsg.length).eql(3000)
+    }, 10000)
+  }).timeout(10000)
 })