Adds tests to check for stream arguments

Author: magne

test/support/rabbit.ts

@@ -1,6 +1,7 @@
 import got from "got"
 import { getTestNodesFromEnv } from "./util"
 import { range } from "../../src/util"
+import { CreateStreamArguments } from "../../src/requests/create_stream_request"
 
 export interface RabbitConnectionResponse {
   name: string
@@ -26,6 +27,7 @@ interface MessageInfoResponse {
   messages_unacknowledged: number
   types: "stream" | "quorum" | "classic"
   node: string
+  arguments?: CreateStreamArguments
 }
 
 interface RabbitPublishersResponse {
@@ -96,6 +98,7 @@ export class Rabbit {
         responseType: "json",
       }
     )
+
     return ret.body
   }
 

test/unit/create_stream.test.ts

@@ -43,6 +43,21 @@ describe("Stream", () => {
       expect(result.name).to.be.eql(streamName)
     })
 
+    it("Should create a new Stream with the given arguments", async () => {
+      const resp = await client.createStream({ stream: streamName, arguments: payload })
+
+      expect(resp).to.be.true
+      const result = await rabbit.getQueueInfo(streamName)
+      expect(result.arguments).to.be.eql({
+        "x-queue-type": "stream",
+        "x-queue-leader-locator": payload["queue-leader-locator"],
+        "x-max-age": payload["max-age"],
+        "x-stream-max-segment-size-bytes": payload["stream-max-segment-size-bytes"],
+        "x-initial-cluster-size": payload["initial-cluster-size"],
+        "x-max-length-bytes": payload["max-length-bytes"],
+      })
+    })
+
     it("Should be idempotent and ignore a duplicate Stream error", async () => {
       await client.createStream({ stream: streamName, arguments: payload })
       const resp = await client.createStream({ stream: streamName, arguments: payload })

test/unit/create_super_stream.test.ts

@@ -47,6 +47,27 @@ describe("Super Stream", () => {
       expect(result.map((r) => r.name)).to.have.members(Array.from(Array(3).keys()).map((n) => `${streamName}-${n}`))
     })
 
+    it("Should create a new Super Stream with 3 partitions by default with the given arguments", async () => {
+      const resp = await client.createSuperStream({ streamName, arguments: payload })
+
+      expect(resp).to.be.true
+      const result = await rabbit.getSuperStreamQueues("%2F", streamName)
+      expect(result.map((r) => r.name)).to.have.members(Array.from(Array(3).keys()).map((n) => `${streamName}-${n}`))
+      await Promise.all(
+        Array.from(Array(3).keys()).map(async (n) => {
+          const queue = await rabbit.getQueueInfo(`${streamName}-${n}`)
+          expect(queue.arguments).to.be.eql({
+            "x-queue-type": "stream",
+            "x-queue-leader-locator": payload["queue-leader-locator"],
+            "x-max-age": payload["max-age"],
+            "x-stream-max-segment-size-bytes": payload["stream-max-segment-size-bytes"],
+            "x-initial-cluster-size": payload["initial-cluster-size"],
+            "x-max-length-bytes": payload["max-length-bytes"],
+          })
+        })
+      )
+    })
+
     it("Should create a new Super Stream with 2 partitions", async () => {
       const resp = await client.createSuperStream({ streamName, arguments: payload }, undefined, 2)