feat: add discard and requeue with annotations

Author: magne

src/delivery_context.ts

@@ -1,9 +1,9 @@
-import { Delivery, Receiver } from "rhea"
+import { Delivery, MessageAnnotations, Receiver } from "rhea"
 
 export interface DeliveryContext {
   accept(): void
-  discard(): void
-  requeue(): void
+  discard(annotations?: MessageAnnotations): void
+  requeue(annotations?: MessageAnnotations): void
 }
 
 export class AmqpDeliveryContext implements DeliveryContext {
@@ -18,15 +18,39 @@ export class AmqpDeliveryContext implements DeliveryContext {
     this.delivery.accept()
   }
 
-  discard(): void {
+  discard(annotations?: MessageAnnotations): void {
     if (this.receiverLink.is_closed()) throw new Error("Receiver link is closed")
+    if (!annotations) {
+      this.delivery.reject()
+      return
+    }
 
-    this.delivery.reject()
+    this.discardWithAnnotations(annotations)
   }
 
-  requeue(): void {
+  private discardWithAnnotations(annotations: MessageAnnotations): void {
+    this.delivery.modified({
+      delivery_failed: true,
+      undeliverable_here: true,
+      message_annotations: annotations,
+    })
+  }
+
+  requeue(annotations?: MessageAnnotations): void {
     if (this.receiverLink.is_closed()) throw new Error("Receiver link is closed")
+    if (!annotations) {
+      this.delivery.release()
+      return
+    }
+
+    this.requeueWithAnnotations(annotations)
+  }
 
-    this.delivery.release()
+  private requeueWithAnnotations(annotations: MessageAnnotations): void {
+    this.delivery.modified({
+      delivery_failed: false,
+      undeliverable_here: false,
+      message_annotations: annotations,
+    })
   }
 }

test/e2e/consumer.test.ts

@@ -228,7 +228,7 @@ describe("Consumer", () => {
     })
   })
 
-  test.skip("consumer can discard a message with annotations in a queue", async () => {
+  test("consumer can discard a message with annotations in a queue", async () => {
     const publisher = await connection.createPublisher({ queue: { name: discardQueueName } })
     const expectedBody = "ciao"
     await publisher.publish(
@@ -240,7 +240,7 @@ describe("Consumer", () => {
     const consumer = await connection.createConsumer({
       queue: { name: discardQueueName },
       messageHandler: (context) => {
-        context.discard()
+        context.discard({ "x-opt-annotation-key": "annotation-value" })
       },
     })
     consumer.start()
@@ -296,7 +296,7 @@ describe("Consumer", () => {
     })
   })
 
-  test.skip("consumer can requeue a message with annotations in a queue", async () => {
+  test("consumer can requeue a message with annotations in a queue", async () => {
     let toRequeue = true
     const messages: Message[] = []
     const consumer = await connection.createConsumer({
@@ -305,7 +305,7 @@ describe("Consumer", () => {
         messages.push(message)
         if (toRequeue) {
           toRequeue = false
-          context.requeue()
+          context.requeue({ "x-opt-annotation-key": "annotation-value" })
           return
         }
         context.accept()
@@ -327,7 +327,7 @@ describe("Consumer", () => {
       expect(messages[0].message_annotations!["x-opt-annotation-key"]).toBeUndefined()
       expect(messages[0].message_annotations!["x-delivery-count"]).toBeUndefined()
       expect(messages[1].message_annotations!["x-opt-annotation-key"]).toEqual("annotation-value")
-      expect(messages[1].message_annotations!["x-delivery-count"]).toEqual("1")
+      expect(messages[1].message_annotations!["x-delivery-count"]).toEqual(1)
     })
-  })
+  }, 15000)
 })