feat: streams support for publishers (#49)

* feat: add annotations to message

* chore: remove console logs

---------

Co-authored-by: magne <[email protected]>

Author: l4mby

conf/enabled_plugins

@@ -1 +1 @@
-[rabbitmq_stream_management,rabbitmq_consistent_hash_exchange].
+[rabbit_stream,rabbitmq_stream_management,rabbitmq_consistent_hash_exchange].

src/consumer.ts

@@ -81,16 +81,12 @@ export class AmqpConsumer implements Consumer {
 
   start() {
     this.receiverLink.on(ReceiverEvents.message, (context: EventContext) => {
-      console.log("message received", context.message?.body)
       if (context.message && context.delivery) {
-        console.log("message accepted")
         try {
           this.params.messageHandler(context.message)
           context.delivery.accept()
-          console.log("message consumed")
         } catch (e) {
           context.delivery.reject({ condition: "Message Handler error", info: e })
-          console.log("message rejected")
         }
       }
     })

src/message.ts

@@ -1,4 +1,4 @@
-import { generate_uuid, Message as RheaMessage } from "rhea"
+import { generate_uuid, MessageAnnotations, Message as RheaMessage } from "rhea"
 import { AmqpEndpoints } from "./link_message_builder.js"
 import { inspect } from "util"
 
@@ -16,6 +16,7 @@ export type DestinationOptions = { exchange: ExchangeOptions } | { queue: QueueO
 type MessageOptions = {
   body: string
   destination?: DestinationOptions
+  annotations?: MessageAnnotations
 }
 
 export function createAmqpMessage(options: MessageOptions): RheaMessage {
@@ -25,10 +26,16 @@ export function createAmqpMessage(options: MessageOptions): RheaMessage {
       body: options.body,
       to: createAddressFrom(options.destination),
       durable: true,
+      message_annotations: options.annotations ?? {},
     }
   }
 
-  return { message_id: generate_uuid(), body: options.body, durable: true }
+  return {
+    message_id: generate_uuid(),
+    body: options.body,
+    durable: true,
+    message_annotations: options.annotations ?? {},
+  }
 }
 
 export function createAddressFrom(options?: DestinationOptions): string | undefined {

test/e2e/publisher.test.ts

@@ -24,6 +24,7 @@ describe("Publisher", () => {
   const queueName2 = "test-queue-2"
   const bindingKey = "test-key"
   const bindingKey2 = "test-key-2"
+  const streamName = "test-stream"
 
   beforeEach(async () => {
     environment = createEnvironment({
@@ -117,4 +118,19 @@ describe("Publisher", () => {
 
     expect(publishResult.outcome).to.eql(OutcomeState.RELEASED)
   })
+
+  test("publish a message to a stream", async () => {
+    const management = connection.management()
+    await management.declareQueue(streamName, { type: "stream" })
+    const publisher = await connection.createPublisher({ queue: { name: streamName } })
+
+    const publishResult = await publisher.publish(
+      createAmqpMessage({
+        body: "Hello World!",
+        annotations: { "x-stream-filter-value": "invoices" },
+      })
+    )
+
+    expect(publishResult.outcome).to.eql(OutcomeState.ACCEPTED)
+  })
 })